Validate old blocks (#268)

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

@@ -0,0 +1,22 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.reader.JsonRpcReader
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.hex.HexQuantity
+import reactor.core.publisher.Mono
+
+class EthereumArchiveBlockNumberReader(
+    private val reader: JsonRpcReader
+) {
+
+    fun readArchiveBlock(): Mono<String> =
+        reader.read(JsonRpcRequest("eth_blockNumber", listOf()))
+            .flatMap(JsonRpcResponse::requireResult)
+            .map {
+                HexQuantity
+                    .from(
+                        String(it).substring(3, it.size - 1).toLong(radix = 16) - 10_000 // this is definitely archive
+                    ).toHex()
+            }
+}

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

@@ -113,7 +113,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
                     .then(Mono.error(TimeoutException("Validation timeout for Peers")))
             )
             .map { count ->
-                val minPeers = options.minPeers ?: 1
+                val minPeers = options.minPeers
                 if (count < minPeers) {
                     UpstreamAvailability.IMMATURE
                 } else {
@@ -137,9 +137,10 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
     fun validateUpstreamSettings(): Boolean {
         return Mono.zip(
             validateChain(),
-            validateCallLimit()
+            validateCallLimit(),
+            validateOldBlocks()
         ).map {
-            it.t1 && it.t2
+            it.t1 && it.t2 && it.t3
         }.block() ?: false
     }
 
@@ -215,6 +216,35 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
             .onErrorReturn(false)
     }
 
+    private fun validateOldBlocks(): Mono<Boolean> {
+        return EthereumArchiveBlockNumberReader(upstream.getIngressReader())
+            .readArchiveBlock()
+            .flatMap {
+                upstream.getIngressReader()
+                    .read(JsonRpcRequest("eth_getBlockByNumber", listOf(it, false)))
+                    .flatMap(JsonRpcResponse::requireResult)
+            }
+            .retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
+                log.warn(
+                    "error during old block retrieving for ${upstream.getId()}, iteration ${ctx.iteration()}",
+                    ctx.exception()
+                )
+            }
+            .map { result ->
+                val receivedResult = result.isNotEmpty() && !Global.nullValue.contentEquals(result)
+                if (!receivedResult) {
+                    log.warn(
+                        "Node ${upstream.getId()} probably is synced incorrectly, it is not possible to get old blocks"
+                    )
+                }
+                true
+            }
+            .onErrorResume {
+                log.warn("Error during old blocks validation", it)
+                Mono.just(true)
+            }
+    }
+
     private fun chainId(): Mono<String> {
         return upstream.getIngressReader()
             .read(JsonRpcRequest("eth_chainId", emptyList()))

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumLabelsDetector.kt

@@ -4,9 +4,9 @@ import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.module.kotlin.readValue
 import io.emeraldpay.dshackle.Global.Companion.objectMapper
 import io.emeraldpay.dshackle.reader.JsonRpcReader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumArchiveBlockNumberReader
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import io.emeraldpay.etherjar.hex.HexQuantity
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 
@@ -39,15 +39,13 @@ class EthereumLabelsDetector(
     }
 
     private fun detectArchiveNode(): Mono<Pair<String, String>> {
-        return reader
+        return EthereumArchiveBlockNumberReader(reader)
-            .read(JsonRpcRequest("eth_blockNumber", listOf()))
+            .readArchiveBlock()
-            .flatMap(JsonRpcResponse::requireResult)
             .flatMap {
-                val blockNum = HexQuantity.from(String(it).substring(3, it.size - 1).toLong(radix = 16) - 10_000) // this is definitely archive
                 reader.read(
                     JsonRpcRequest(
                         "eth_getBalance",
-                        listOf("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", blockNum.toHex())
+                        listOf("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", it)
                     )
                 )
             }

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy

@@ -101,7 +101,7 @@ class TrackEthereumTxSpec extends Specification {
         def apiMock = TestingCommons.api()
         def upstreamMock = TestingCommons.upstream(apiMock)
         MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel())
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         apiMock.answer("eth_getTransactionByHash", [txId], txJson)
         apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson)
@@ -192,7 +192,7 @@ class TrackEthereumTxSpec extends Specification {
         def apiMock = TestingCommons.api()
         def upstreamMock = TestingCommons.upstream(apiMock)
         MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel())
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM__MAINNET, Instant.now(), TransactionId.from(txId), 6)
         def block = new BlockContainer(
@@ -214,7 +214,7 @@ class TrackEthereumTxSpec extends Specification {
         def apiMock = TestingCommons.api()
         def upstreamMock = TestingCommons.upstream(apiMock)
         MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel())
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM__MAINNET, Instant.now(), TransactionId.from(txId), 6)
         def block = new BlockContainer(
@@ -293,7 +293,7 @@ class TrackEthereumTxSpec extends Specification {
         }
         MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, multi)
 
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel())
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         apiMock.answerOnce("eth_getTransactionByHash", [txId], null)
         apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted)

src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy

@@ -30,9 +30,9 @@ class ConfiguredUpstreamsSpec extends Specification {
                 Executors.newFixedThreadPool(1),
                 ChainsConfig.default(),
                 GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
                 null,
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
         )
         def methods = new UpstreamsConfig.Methods(
                 [
@@ -62,9 +62,9 @@ class ConfiguredUpstreamsSpec extends Specification {
                 Executors.newFixedThreadPool(1),
                 ChainsConfig.default(),
                 GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
                 null,
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
         )
         def methods = new UpstreamsConfig.Methods(
                 [
@@ -93,9 +93,9 @@ class ConfiguredUpstreamsSpec extends Specification {
                 Executors.newFixedThreadPool(1),
                 ChainsConfig.default(),
                 GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
                 null,
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
         )
         expect:
         configurer.getHash(node, src) == expected
@@ -119,9 +119,9 @@ class ConfiguredUpstreamsSpec extends Specification {
                 Executors.newFixedThreadPool(1),
                 ChainsConfig.default(),
                 GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
                 null,
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
         )
         when:
         def h1 = configurer.getHash(null, "hohoho")
@@ -150,9 +150,9 @@ class ConfiguredUpstreamsSpec extends Specification {
                 Executors.newFixedThreadPool(1),
                 ChainsConfig.default(),
                 GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
                 null,
-                Schedulers.parallel(),
+                Schedulers.boundedElastic(),
         )
         def methodsGroup = new UpstreamsConfig.MethodGroups(
                 ["filter"] as Set,

src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy

@@ -48,7 +48,7 @@ class MultistreamHolderMock implements MultistreamHolder {
                 } else if (up instanceof EthereumLikeRpcUpstream) {
                     upstreams[chain] = new EthereumPosMultiStream(
                             chain, [up as EthereumLikeRpcUpstream], Caches.default(),
-                            Schedulers.parallel(), TestingCommons.tracerMock()
+                            Schedulers.boundedElastic(), TestingCommons.tracerMock()
                     )
                 } else {
                     throw new IllegalArgumentException("Unsupported upstream type ${up.class}")
@@ -97,7 +97,7 @@ class MultistreamHolderMock implements MultistreamHolder {
         Head customHead = null
 
         EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumLikeRpcUpstream> upstreams, @NotNull Caches caches) {
-            super(chain, upstreams, caches, Schedulers.parallel(), new BraveTracer(null, null, null))
+            super(chain, upstreams, caches, Schedulers.boundedElastic(), new BraveTracer(null, null, null))
         }
 
         EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumLikeRpcUpstream> upstreams) {

src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy

@@ -91,7 +91,7 @@ class TestingCommons {
     }
 
     static Multistream multistream(EthereumPosRpcUpstreamMock up) {
-        return new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, [up], Caches.default(), Schedulers.parallel(), tracerMock()).tap {
+        return new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, [up], Caches.default(), Schedulers.boundedElastic(), tracerMock()).tap {
             start()
         }
     }
@@ -112,11 +112,11 @@ class TestingCommons {
     }
 
     static Multistream multistreamWithoutUpstreams(Chain chain) {
-        return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), Schedulers.parallel(), tracerMock())
+        return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), tracerMock())
     }
 
     static Multistream multistreamClassicWithoutUpstreams(Chain chain) {
-        return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.parallel(), tracerMock())
+        return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), tracerMock())
     }
 
     static FileResolver fileResolver() {

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

@@ -138,7 +138,7 @@ class AbstractHeadSpec extends Specification {
                 BlockContainer getHead() {
                     return null
                 }
-            }, Schedulers.parallel(),   new BlockValidator.AlwaysValid() , 100_000)
+            }, Schedulers.boundedElastic(),   new BlockValidator.AlwaysValid() , 100_000)
         }
     }
 }

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

@@ -58,8 +58,8 @@ class FilteredApisSpec extends Specification {
                     httpFactory,
                     new MostWorkForkChoice(),
                     BlockValidator.ALWAYS_VALID,
-                    Schedulers.parallel(),
+                    Schedulers.boundedElastic(),
-                    Schedulers.parallel()
+                    Schedulers.boundedElastic()
             )
             new EthereumLikeRpcUpstream(
                     "test",

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

@@ -37,7 +37,7 @@ class MergedHeadSpec extends Specification {
         }
 
         when:
-        def merged = new MergedHead([head1, head2, head3], new MostWorkForkChoice(), Schedulers.parallel())
+        def merged = new MergedHead([head1, head2, head3], new MostWorkForkChoice(), Schedulers.boundedElastic())
         merged.start()
 
         then:
@@ -46,14 +46,14 @@ class MergedHeadSpec extends Specification {
 
     class TestHead1 extends AbstractHead {
         TestHead1() {
-            super(new MostWorkForkChoice(), Schedulers.parallel(), new BlockValidator.AlwaysValid(), 100_000)
+            super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000)
         }
     }
 
     class TestHead2 extends AbstractHead implements Lifecycle {
 
         TestHead2() {
-            super(new MostWorkForkChoice(), Schedulers.parallel(), new BlockValidator.AlwaysValid(), 100_000)
+            super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000)
         }
 
         @Override

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

@@ -52,7 +52,7 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
         def up2 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"]))
-        def aggr = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, [up1, up2], Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
+        def aggr = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, [up1, up2], Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock())
         when:
         aggr.onUpstreamsUpdated()
         def act = aggr.getMethods()
@@ -183,7 +183,7 @@ class MultistreamSpec extends Specification {
         def up1 = TestingCommons.upstream("test-1", "internal")
         def up2 = TestingCommons.upstream("test-2", "external")
         def up3 = TestingCommons.upstream("test-3", "external")
-        def multistream = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, [up1, up2, up3], Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
+        def multistream = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, [up1, up2, up3], Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock())
 
         expect:
         multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock)
@@ -252,7 +252,7 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"]))
         def up2 = new EthereumPosRpcUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
-        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumLikeRpcUpstream>(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
+        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumLikeRpcUpstream>(), Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock())
         when:
         ms.onUpstreamChange(
                 new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)
@@ -305,7 +305,7 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"]))
         def up2 = new EthereumPosRpcUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
-        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumLikeRpcUpstream>(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
+        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumLikeRpcUpstream>(), Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock())
         def head1 = createBlock(250, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448401b")
         def head2 = createBlock(270, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448402b")
         def head3 = createBlock(100, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448412b")
@@ -340,7 +340,7 @@ class MultistreamSpec extends Specification {
     class TestEthereumPosMultistream extends EthereumPosMultiStream {
 
         TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumLikeRpcUpstream> upstreams, @NotNull Caches caches) {
-            super(chain, upstreams, caches, Schedulers.parallel(), TestingCommons.tracerMock())
+            super(chain, upstreams, caches, Schedulers.boundedElastic(), TestingCommons.tracerMock())
         }
 
         @NotNull

src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy

@@ -86,7 +86,7 @@ class BitcoinRpcHeadSpec extends Specification {
             _ * read(new JsonRpcRequest("getblock", [hash1])) >> Mono.just(new JsonRpcResponse(block1.bytes, null))
             _ * read(new JsonRpcRequest("getblock", [hash2])) >> Mono.just(new JsonRpcResponse(block2.bytes, null))
         }
-        BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(), Duration.ofMillis(200), Schedulers.parallel())
+        BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(), Duration.ofMillis(200), Schedulers.boundedElastic())
 
         when:
         def act = head.flux.take(2)

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

@@ -32,7 +32,7 @@ import java.time.Instant
 
 class DefaultEthereumHeadSpec extends Specification {
 
-    DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.parallel())
+    DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.boundedElastic())
     ObjectMapper objectMapper = Global.objectMapper
     BlockHash parent = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915210")
 

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

@@ -38,7 +38,7 @@ class EnrichedMergedHeadSpec extends Specification {
 
         def api = new ApiReaderMock()
         when:
-        def merged = new EnrichedMergedHead([head1, head2], head3, Schedulers.parallel(), new BlockReader(api))
+        def merged = new EnrichedMergedHead([head1, head2], head3, Schedulers.boundedElastic(), new BlockReader(api))
         merged.start()
 
         then:
@@ -56,7 +56,7 @@ class EnrichedMergedHeadSpec extends Specification {
             _ * getFlux() >> Flux.just(block)
         }
         when:
-        def merge = new EnrichedMergedHead([], head, Schedulers.parallel(), new BlockReader(api))
+        def merge = new EnrichedMergedHead([], head, Schedulers.boundedElastic(), new BlockReader(api))
 
         then:
         StepVerifier.create(merge.getFlux())
@@ -81,7 +81,7 @@ class EnrichedMergedHeadSpec extends Specification {
             _ * getFlux() >> Flux.just(enrichedBlock)
         }
         when:
-        def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.parallel(), new BlockReader(new ApiReaderMock()))
+        def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.boundedElastic(), new BlockReader(new ApiReaderMock()))
         then:
         StepVerifier.create(merge.getFlux())
             .then { merge.start() }
@@ -106,7 +106,7 @@ class EnrichedMergedHeadSpec extends Specification {
             _ * getFlux() >> sourceSink.asFlux()
         }
         when:
-        def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.parallel(), new BlockReader(new ApiReaderMock()))
+        def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.boundedElastic(), new BlockReader(new ApiReaderMock()))
         then:
         StepVerifier.create(merge.getFlux())
             .then { merge.start() }
@@ -133,7 +133,7 @@ class EnrichedMergedHeadSpec extends Specification {
             answer("eth_getBlockByHash", [block.hash.toHexWithPrefix(), false], enrichedBlock.toBlock())
         }
         when:
-        def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.parallel(), new BlockReader(api))
+        def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.boundedElastic(), new BlockReader(api))
         then:
         StepVerifier.create(merge.getFlux())
             .then { merge.start() }
@@ -164,7 +164,7 @@ class EnrichedMergedHeadSpec extends Specification {
     class TestHead extends AbstractHead implements Lifecycle {
 
         TestHead() {
-            super(new MostWorkForkChoice(), Schedulers.parallel(), new BlockValidator.AlwaysValid(), 100_000)
+            super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000)
         }
 
         @Override

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

@@ -26,7 +26,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read empty logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([:])
 
@@ -37,7 +37,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read single address logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "0x829bd824b016326a401d083b33d092293333a830"
@@ -62,7 +62,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "ignores invalid address for logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "829bd824b016326a401d083b33d092293333a830"
@@ -75,7 +75,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read multi address logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: ["0x829bd824b016326a401d083b33d092293333a830", "0x401d083b33d092293333a83829bd824b016326a0"]
@@ -91,7 +91,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read single topic logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
@@ -116,7 +116,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read invalid topic for request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: [
@@ -134,7 +134,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read multi topic logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: [
@@ -153,7 +153,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read full logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",

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

@@ -263,20 +263,26 @@ class EthereumUpstreamValidatorSpec extends Specification {
         act == UNAVAILABLE
     }
 
-    def "Doesnt validate settings when disabled"() {
+    def "Doesnt validate chan and callLimit when disabled"() {
         setup:
         def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validateCalllimit = false
             it.validateChain = false
         }.buildOptions()
-        def up = Mock(EthereumLikeUpstream)
+        def up = Mock(EthereumLikeUpstream) {
+            2 * getIngressReader() >>
+                    Mock(Reader) {
+                        1 * read(new JsonRpcRequest("eth_blockNumber", [])) >> Mono.just(new JsonRpcResponse('"0x10ff9be"'.getBytes(), null))
+                        1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x10fd2ae", false])) >>
+                                Mono.just(new JsonRpcResponse('"result"'.getBytes(), null))
+                    }
+        }
         def validator = new EthereumUpstreamValidator(ETHEREUM__MAINNET, up, options)
 
         when:
         def act = validator.validateUpstreamSettings()
         then:
         act
-        0 * up.getIngressReader()
     }
 
     def "Upstream is valid if not error from call limit check"() {
@@ -285,11 +291,14 @@ class EthereumUpstreamValidatorSpec extends Specification {
             it.validateChain = false
         }.buildOptions()
         def up = Mock(EthereumLikeRpcUpstream) {
-            1 * getIngressReader() >> Mock(Reader) {
+            3 * getIngressReader() >> Mock(Reader) {
                 1 * read(new JsonRpcRequest("eth_call", [new TransactionCallJson(
                         Address.from("0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96"),
                         HexData.from("0xd8a26e3a0000000000000000000000000000000000000000000000000000000000030ce0")
                 ), "latest"])) >> Mono.just(new JsonRpcResponse("0x00000000000000000000".getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_blockNumber", [])) >> Mono.just(new JsonRpcResponse('"0x10ff9be"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x10fd2ae", false])) >>
+                        Mono.just(new JsonRpcResponse('"result"'.getBytes(), null))
             }
         }
         def validator = new EthereumUpstreamValidator(ETHEREUM__MAINNET, up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
@@ -306,11 +315,14 @@ class EthereumUpstreamValidatorSpec extends Specification {
             it.validateChain = false
         }.buildOptions()
         def up = Mock(EthereumLikeRpcUpstream) {
-            1 * getIngressReader() >> Mock(Reader) {
+            3 * getIngressReader() >> Mock(Reader) {
                 1 * read(new JsonRpcRequest("eth_call", [new TransactionCallJson(
                         Address.from("0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96"),
                         HexData.from("0xd8a26e3a0000000000000000000000000000000000000000000000000000000000030ce0")
                 ), "latest"])) >> Mono.just(new JsonRpcResponse(null, new JsonRpcError(1, "Too long")))
+                1 * read(new JsonRpcRequest("eth_blockNumber", [])) >> Mono.just(new JsonRpcResponse('"0x10ff9be"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x10fd2ae", false])) >>
+                        Mono.just(new JsonRpcResponse('"result"'.getBytes(), null))
             }
         }
         def validator = new EthereumUpstreamValidator(ETHEREUM__MAINNET, up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
@@ -327,9 +339,12 @@ class EthereumUpstreamValidatorSpec extends Specification {
             it.validateCalllimit = false
         }.buildOptions()
         def up = Mock(EthereumLikeRpcUpstream) {
-            2 * getIngressReader() >> Mock(Reader) {
+            4 * getIngressReader() >> Mock(Reader) {
                 1 * read(new JsonRpcRequest("eth_chainId", emptyList())) >> Mono.just(new JsonRpcResponse('"0x1"'.getBytes(), null))
                 1 * read(new JsonRpcRequest("net_version", emptyList())) >> Mono.just(new JsonRpcResponse('"1"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_blockNumber", [])) >> Mono.just(new JsonRpcResponse('"0x10ff9be"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x10fd2ae", false])) >>
+                        Mono.just(new JsonRpcResponse('"result"'.getBytes(), null))
             }
         }
         def validator = new EthereumUpstreamValidator(ETHEREUM__MAINNET, up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
@@ -346,9 +361,12 @@ class EthereumUpstreamValidatorSpec extends Specification {
             it.validateCalllimit = false
         }.buildOptions()
         def up = Mock(EthereumLikeRpcUpstream) {
-            2 * getIngressReader() >> Mock(Reader) {
+            4 * getIngressReader() >> Mock(Reader) {
                 1 * read(new JsonRpcRequest("eth_chainId", emptyList())) >> Mono.just(new JsonRpcResponse('"0x1"'.getBytes(), null))
                 1 * read(new JsonRpcRequest("net_version", emptyList())) >> Mono.just(new JsonRpcResponse('"1"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_blockNumber", [])) >> Mono.just(new JsonRpcResponse('"0x10ff9be"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x10fd2ae", false])) >>
+                        Mono.just(new JsonRpcResponse('"result"'.getBytes(), null))
             }
         }
         def validator = new EthereumUpstreamValidator(OPTIMISM__MAINNET, up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
@@ -363,13 +381,16 @@ class EthereumUpstreamValidatorSpec extends Specification {
         setup:
         def options = UpstreamsConfig.PartialOptions.getDefaults().buildOptions()
         def up = Mock(EthereumLikeRpcUpstream) {
-            3 * getIngressReader() >> Mock(Reader) {
+            5 * getIngressReader() >> Mock(Reader) {
                 1 * read(new JsonRpcRequest("eth_chainId", emptyList())) >> Mono.just(new JsonRpcResponse('"0x1"'.getBytes(), null))
                 1 * read(new JsonRpcRequest("net_version", emptyList())) >> Mono.just(new JsonRpcResponse('"1"'.getBytes(), null))
                 1 * read(new JsonRpcRequest("eth_call", [new TransactionCallJson(
                         Address.from("0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96"),
                         HexData.from("0xd8a26e3a0000000000000000000000000000000000000000000000000000000000030ce0")
                 ), "latest"])) >> Mono.just(new JsonRpcResponse("0x00000000000000000000".getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_blockNumber", [])) >> Mono.just(new JsonRpcResponse('"0x10ff9be"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x10fd2ae", false])) >>
+                        Mono.just(new JsonRpcResponse('"result"'.getBytes(), null))
             }
         }
         def validator = new EthereumUpstreamValidator(ETHEREUM__MAINNET, up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
@@ -384,13 +405,16 @@ class EthereumUpstreamValidatorSpec extends Specification {
         setup:
         def options = UpstreamsConfig.PartialOptions.getDefaults().buildOptions()
         def up = Mock(EthereumLikeRpcUpstream) {
-            3 * getIngressReader() >> Mock(Reader) {
+            5 * getIngressReader() >> Mock(Reader) {
                 1 * read(new JsonRpcRequest("eth_chainId", emptyList())) >> Mono.just(new JsonRpcResponse(null, new JsonRpcError(1, "Too long")))
                 1 * read(new JsonRpcRequest("net_version", emptyList())) >> Mono.just(new JsonRpcResponse(null, new JsonRpcError(1, "Too long")))
                 1 * read(new JsonRpcRequest("eth_call", [new TransactionCallJson(
                         Address.from("0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96"),
                         HexData.from("0xd8a26e3a0000000000000000000000000000000000000000000000000000000000030ce0")
                 ), "latest"])) >> Mono.just(new JsonRpcResponse(null, new JsonRpcError(1, "Too long")))
+                1 * read(new JsonRpcRequest("eth_blockNumber", [])) >> Mono.just(new JsonRpcResponse('"0x10ff9be"'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x10fd2ae", false])) >>
+                        Mono.just(new JsonRpcResponse('"result"'.getBytes(), null))
             }
         }
         def validator = new EthereumUpstreamValidator(ETHEREUM__MAINNET, up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
@@ -401,4 +425,5 @@ class EthereumUpstreamValidatorSpec extends Specification {
         !act
     }
 
+
 }

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

@@ -66,7 +66,7 @@ class EthereumWsHeadSpec extends Specification {
             1 * it.connectionInfoFlux() >> Flux.empty()
         }
 
-        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.parallel(), Schedulers.parallel())
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.boundedElastic(), Schedulers.boundedElastic())
 
         when:
         def act = head.listenNewHeads().blockFirst()
@@ -107,7 +107,7 @@ class EthereumWsHeadSpec extends Specification {
             ]
         }
 
-        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel())
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic())
 
         when:
         def act = head.getFlux()
@@ -161,7 +161,7 @@ class EthereumWsHeadSpec extends Specification {
             ]
         }
 
-        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel())
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic())
 
         when:
         def act = head.getFlux()
@@ -201,7 +201,7 @@ class EthereumWsHeadSpec extends Specification {
             ]
         }
 
-        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel())
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic())
 
         when:
         def act = head.getFlux()
@@ -240,7 +240,7 @@ class EthereumWsHeadSpec extends Specification {
             ]
         }
 
-        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel())
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic())
 
         when:
         def act = head.getFlux()
@@ -293,7 +293,7 @@ class EthereumWsHeadSpec extends Specification {
             ]
         }
 
-        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel())
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic())
 
         when:
         def act = head.getFlux()

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

@@ -40,7 +40,7 @@ class WsConnectionImplRealSpec extends Specification {
                         Chain.ETHEREUM__MAINNET,
                         "ws://localhost:${port}".toURI(),
                         "http://localhost:${port}".toURI(),
-                        Schedulers.parallel()
+                        Schedulers.boundedElastic()
                 )
         ).create(null).getConnection()
     }
@@ -120,7 +120,7 @@ class WsConnectionImplRealSpec extends Specification {
                         Chain.ETHEREUM__MAINNET,
                         "ws://localhost:${port}".toURI(),
                         "http://localhost:${port}".toURI(),
-                        Schedulers.parallel()
+                        Schedulers.boundedElastic()
                 )
         ).create(up).getConnection()
         when:

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

@@ -41,7 +41,7 @@ class WsConnectionImplSpec extends Specification {
                         Chain.ETHEREUM__MAINNET,
                         new URI("http://localhost"),
                         new URI("http://localhost"),
-                        Schedulers.parallel()
+                        Schedulers.boundedElastic()
                 )
         )
         def apiMock = TestingCommons.api()
@@ -76,7 +76,7 @@ class WsConnectionImplSpec extends Specification {
                         Chain.ETHEREUM__MAINNET,
                         new URI("http://localhost"),
                         new URI("http://localhost"),
-                        Schedulers.parallel()
+                        Schedulers.boundedElastic()
                 )
         )
         def apiMock = TestingCommons.api()
@@ -109,7 +109,7 @@ class WsConnectionImplSpec extends Specification {
                         Chain.ETHEREUM__MAINNET,
                         new URI("http://localhost"),
                         new URI("http://localhost"),
-                        Schedulers.parallel()
+                        Schedulers.boundedElastic()
                 )
         )
         def apiMock = TestingCommons.api()

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy

@@ -38,7 +38,7 @@ class ConnectBlockUpdatesSpec extends Specification {
 
     def "Extracts updates"() {
         setup:
-        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel())
+        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic())
         def block = BlockContainer.from(new BlockJson<TransactionRefJson>().tap {
             hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c")
             number = 13412871
@@ -72,7 +72,7 @@ class ConnectBlockUpdatesSpec extends Specification {
 
     def "Produce DROP updates for replaced block"() {
         setup:
-        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel())
+        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic())
         def block = BlockContainer.from(new BlockJson<TransactionRefJson>().tap {
             hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c")
             number = 13412871
@@ -106,7 +106,7 @@ class ConnectBlockUpdatesSpec extends Specification {
 
     def "Gets prev version if available"() {
         setup:
-        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel())
+        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic())
         def block1 = BlockContainer.from(new BlockJson<TransactionRefJson>().tap {
             hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c")
             number = 13412871
@@ -170,7 +170,7 @@ class ConnectBlockUpdatesSpec extends Specification {
 
     def "Marks old txes as dropped before producing a new version of same block"() {
         setup:
-        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel())
+        def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic())
         def block1 = BlockContainer.from(new BlockJson<TransactionRefJson>().tap {
             hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c")
             number = 13412871
@@ -235,7 +235,7 @@ class ConnectBlockUpdatesSpec extends Specification {
         def up = Mock(EthereumMultistream) {
             1 * getEnrichedHead(Selector.empty) >> head
         }
-        def connectBlockUpdates = new ConnectBlockUpdates(up, Schedulers.parallel())
+        def connectBlockUpdates = new ConnectBlockUpdates(up, Schedulers.boundedElastic())
 
         when:
         def a1 = connectBlockUpdates.connect()

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy

@@ -91,7 +91,7 @@ class ConnectLogsSpec extends Specification {
 
     def "Filter is empty"() {
         setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel())
+        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic())
         when:
         def input = Flux.fromIterable([
                 log1, log2, log3, log4
@@ -109,7 +109,7 @@ class ConnectLogsSpec extends Specification {
 
     def "Filter by address"() {
         setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel())
+        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic())
         when:
         def input = Flux.fromIterable([
                 log1, log2
@@ -124,7 +124,7 @@ class ConnectLogsSpec extends Specification {
 
     def "Filter by topic"() {
         setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel())
+        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic())
         when:
         def input = Flux.fromIterable([
                 log1, log2, log3, log4
@@ -141,7 +141,7 @@ class ConnectLogsSpec extends Specification {
 
     def "Filter by address and topic"() {
         setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel())
+        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic())
         when:
         def input = Flux.fromIterable([
                 log1, log2, log3, log4

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy

@@ -21,7 +21,7 @@ class ConnectNewHeadsSpec extends Specification {
         def up = Mock(EthereumMultistream) {
             1 * getHead(Selector.empty) >> head
         }
-        ConnectNewHeads connectNewHeads = new ConnectNewHeads(up, Schedulers.parallel())
+        ConnectNewHeads connectNewHeads = new ConnectNewHeads(up, Schedulers.boundedElastic())
         when:
         def act1 = connectNewHeads.connect(Selector.empty)
         def act2 = connectNewHeads.connect(Selector.empty)

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy

@@ -91,7 +91,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 )
             }
         })
-        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel())
+        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic())
         upstream.setLag(0)
         upstream.update(
                 BlockchainOuterClass.DescribeChain.newBuilder()
@@ -162,7 +162,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 }).start()
             }
         })
-        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM__MAINNET, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM__MAINNET, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel())
+        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM__MAINNET, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM__MAINNET, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic())
         upstream.setLag(0)
         upstream.update(
                 BlockchainOuterClass.DescribeChain.newBuilder()
@@ -234,7 +234,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 finished.complete(true)
             }
         })
-        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel())
+        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic())
         upstream.setLag(0)
         upstream.update(
                 BlockchainOuterClass.DescribeChain.newBuilder()
@@ -268,7 +268,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
             void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
             }
         })
-        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel())
+        def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic())
         upstream.setLag(0)
         upstream.setStatus(UpstreamAvailability.OK)
         when:

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy

@@ -70,7 +70,7 @@ class GrpcHeadSpec extends Specification {
                 convert,
                 null,
                 new MostWorkForkChoice(),
-                Schedulers.parallel()
+                Schedulers.boundedElastic()
         )
         when:
         def act = head.getFlux()
@@ -137,7 +137,7 @@ class GrpcHeadSpec extends Specification {
                 convert,
                 null,
                 new MostWorkForkChoice(),
-                Schedulers.parallel()
+                Schedulers.boundedElastic()
         )
         when:
         def act = head.getFlux()