From 363af28bfcbf247216ff1c8999f1619957aa9caf Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Fri, 4 Aug 2023 15:21:04 +0400 Subject: [PATCH] Validate old blocks (#268) --- .../EthereumArchiveBlockNumberReader.kt | 22 ++++++++++ .../ethereum/EthereumUpstreamValidator.kt | 36 ++++++++++++++-- .../subscribe/EthereumLabelsDetector.kt | 10 ++--- .../dshackle/rpc/TrackEthereumTxSpec.groovy | 8 ++-- .../startup/ConfiguredUpstreamsSpec.groovy | 20 ++++----- .../test/MultistreamHolderMock.groovy | 4 +- .../dshackle/test/TestingCommons.groovy | 6 +-- .../dshackle/upstream/AbstractHeadSpec.groovy | 2 +- .../dshackle/upstream/FilteredApisSpec.groovy | 4 +- .../dshackle/upstream/MergedHeadSpec.groovy | 6 +-- .../dshackle/upstream/MultistreamSpec.groovy | 10 ++--- .../bitcoin/BitcoinRpcHeadSpec.groovy | 2 +- .../ethereum/DefaultEthereumHeadSpec.groovy | 2 +- .../ethereum/EnrichedMergedHeadSpec.groovy | 12 +++--- .../EthereumEgressSubscriptionSpec.groovy | 16 +++---- .../EthereumUpstreamValidatorSpec.groovy | 43 +++++++++++++++---- .../ethereum/EthereumWsHeadSpec.groovy | 12 +++--- .../ethereum/WsConnectionImplRealSpec.groovy | 4 +- .../ethereum/WsConnectionImplSpec.groovy | 6 +-- .../subscribe/ConnectBlockUpdatesSpec.groovy | 10 ++--- .../ethereum/subscribe/ConnectLogsSpec.groovy | 8 ++-- .../subscribe/ConnectNewHeadsSpec.groovy | 2 +- .../grpc/EthereumGrpcUpstreamSpec.groovy | 8 ++-- .../upstream/grpc/GrpcHeadSpec.groovy | 4 +- 24 files changed, 166 insertions(+), 91 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumArchiveBlockNumberReader.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumArchiveBlockNumberReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumArchiveBlockNumberReader.kt new file mode 100644 index 00000000..97a50d44 --- /dev/null +++ b/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 = + 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() + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt index 07cb3fc4..1dc509b4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/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 { + 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 { return upstream.getIngressReader() .read(JsonRpcRequest("eth_chainId", emptyList())) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumLabelsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumLabelsDetector.kt index 3277c59d..1f20f63d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumLabelsDetector.kt +++ b/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> { - return reader - .read(JsonRpcRequest("eth_blockNumber", listOf())) - .flatMap(JsonRpcResponse::requireResult) + return EthereumArchiveBlockNumberReader(reader) + .readArchiveBlock() .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) ) ) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index 35d5e09d..f4b81ebb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index 99cb4750..eb392af0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/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, diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index e85a2e4c..06be7160 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/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 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 upstreams) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 9af3eb3e..9619b918 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/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() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy index 3aad3e43..aa980a7a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy +++ b/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) } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index d0bca0fa..397b827d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/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.parallel() + Schedulers.boundedElastic(), + Schedulers.boundedElastic() ) new EthereumLikeRpcUpstream( "test", diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy index cbb29802..3d84b519 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy +++ b/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 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index 6a0113ec..d73c468d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/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(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) + def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList(), 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(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) + def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList(), 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 upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, Schedulers.parallel(), TestingCommons.tracerMock()) + super(chain, upstreams, caches, Schedulers.boundedElastic(), TestingCommons.tracerMock()) } @NotNull diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy index c0ecccc1..45a1cebf 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy +++ b/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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy index 98a7ef48..76514e25 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy +++ b/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") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy index b61a6ae9..bbeffef3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy +++ b/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 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy index 59bc7a5a..0a384473 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy +++ b/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", diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy index d437790b..1bf07265 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy +++ b/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 } + } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy index 58a33a40..8cbf8f22 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy +++ b/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() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy index daaf4fa1..e9a0b79a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy +++ b/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: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy index 180d853c..edaf98b4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy +++ b/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() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy index a2b85e6b..3e567eec 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy +++ b/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().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().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().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().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() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy index 208e6d3c..0de35e85 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy +++ b/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 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy index bda68a55..998640c3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy +++ b/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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index 0b17d0f8..26587d6a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/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 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: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index 8ee22c2d..cfb375a4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/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()