From 706f472355d329460f448d356bc45b43755f39d4 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 6 Sep 2019 00:15:10 -0400 Subject: [PATCH] solution: upstream apis as a controller Publisher --- .../dshackle/reader/BlockApiReader.kt | 3 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 33 ++-- .../emeraldpay/dshackle/rpc/TrackAddress.kt | 2 +- .../io/emeraldpay/dshackle/rpc/TrackTx.kt | 4 +- .../dshackle/upstream/AggregatedUpstream.kt | 3 +- .../emeraldpay/dshackle/upstream/ApiSource.kt | 26 +++ .../dshackle/upstream/ChainUpstreams.kt | 12 +- .../dshackle/upstream/FilteredApis.kt | 102 ++++++++++ .../dshackle/upstream/FilteringApiIterator.kt | 59 ------ .../emeraldpay/dshackle/upstream/Upstream.kt | 3 +- .../dshackle/upstream/UpstreamValidator.kt | 32 +-- .../upstream/ethereum/EthereumUpstream.kt | 5 +- .../dshackle/upstream/grpc/GrpcUpstream.kt | 6 +- .../dshackle/rpc/NativeCallSpec.groovy | 2 +- .../dshackle/test/EthereumApiStub.groovy | 76 ++++++++ .../dshackle/upstream/FilteredApisSpec.groovy | 182 ++++++++++++++++++ .../upstream/FilteringApiIteratorSpec.groovy | 96 --------- 17 files changed, 443 insertions(+), 203 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy delete mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt index 1e4bec7a..a5bdcd90 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt @@ -32,7 +32,8 @@ class BlockApiReader( override fun read(key: BlockHash): Mono> { return Mono.just(key) .flatMap { - upstream.getApi(Selector.empty).executeAndConvert(Commands.eth().getBlock(it)) + upstream.getApi(Selector.empty) + .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(it)) } }.repeatWhenEmpty { n -> Repeat.times(3) .exponentialBackoff(Duration.ofMillis(100), Duration.ofMillis(500)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index b65e79d0..cdcdf0dd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -19,7 +19,6 @@ import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.* -import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.grpc.Chain @@ -30,9 +29,6 @@ import org.springframework.stereotype.Service import reactor.core.publisher.* import reactor.util.function.Tuples import java.lang.Exception -import java.time.Duration -import java.util.concurrent.atomic.AtomicInteger -import java.util.function.Predicate @Service class NativeCall( @@ -122,24 +118,27 @@ class NativeCall( } fun executeOnRemote(ctx: CallContext): Mono> { - val all = ctx.getApis().toFlux().share() - //execute on the first API immediately, and then make a delay between each call to not overload upstreams - val immediate = Flux.from(all).take(1) - val repeatControl = EmitterProcessor.create() - val retries = Flux.from(all).skip(1) - .zipWith(repeatControl.delayElements(Duration.ofMillis(200))) //manages when need another call, make delay for at least of 200ms between calls - .map { it.t1 } - - return Flux.concat(immediate, retries) + val apis = ctx.getApis() + apis.request(1) + var failures = 0 + return Flux.from(apis) .flatMap { api -> api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) } } - .retry(3) + .retry { + failures++ + if (failures <= 3) { + apis.request(1) + true + } else { + false + } + } .reduce(ctx.callQuorum, {res, a -> if (res.record(a.t1, a.t2)) { - repeatControl.onComplete() + apis.resolve() } else { - repeatControl.onNext(true) + apis.request(1) } res }) @@ -181,7 +180,7 @@ class NativeCall( return CallContext(id, upstream, matcher, callQuorum, payload) } - fun getApis(): Iterator { + fun getApis(): ApiSource { return upstream.getApis(matcher) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index a6f0cb1f..e89d190a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -168,7 +168,7 @@ class TrackAddress( fun getBalance(addr: SimpleAddress): Mono { val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(Exception("Unsupported chain: ${addr.chain}")) return up.getApi(Selector.empty) - .executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) + .flatMap { api -> api.executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) } .timeout(Duration.ofSeconds(15)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index d8708f30..0e97fe27 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -208,7 +208,7 @@ class TrackTx( val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}")) return upstream.getApi(Selector.empty) - .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) + .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) } .map { block -> setBlockDetails(tx, block) }.doOnError { t -> @@ -249,7 +249,7 @@ class TrackTx( val initialStatus = tx.status val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}")) val execution = upstream.getApi(Selector.empty) - .executeAndConvert(Commands.eth().getTransaction(tx.txid)) + .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) } return execution .flatMap { updateFromBlock(upstream, tx, it) } .doOnError { t -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt index fde1e178..620f05a7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt @@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson +import org.reactivestreams.Publisher import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux @@ -51,7 +52,7 @@ abstract class AggregatedUpstream( abstract fun getAll(): List abstract fun addUpstream(upstream: Upstream) - abstract fun getApis(matcher: Selector.Matcher): Iterator + abstract fun getApis(matcher: Selector.Matcher): ApiSource fun onUpstreamsUpdated() { reconfigLock.withLock { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt new file mode 100644 index 00000000..4024d2ac --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt @@ -0,0 +1,26 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi +import org.reactivestreams.Publisher + +interface ApiSource: Publisher { + + fun resolve() + fun request(tries: Int) + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 93f067c6..8ead0d68 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -24,6 +24,7 @@ import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable +import reactor.core.publisher.Mono import java.lang.IllegalStateException import java.time.Duration @@ -116,16 +117,19 @@ open class ChainUpstreams ( } } - override fun getApis(matcher: Selector.Matcher): Iterator { + override fun getApis(matcher: Selector.Matcher): ApiSource { val i = seq++ if (seq >= Int.MAX_VALUE / 2) { seq = 0 } - return FilteringApiIterator(upstreams, i, matcher) + return FilteredApis(upstreams, matcher, i) } - override fun getApi(matcher: Selector.Matcher): DirectEthereumApi { - return getApis(matcher).next() + override fun getApi(matcher: Selector.Matcher): Mono { + val apis = getApis(matcher) + apis.request(1) + return Mono.from(apis) + .switchIfEmpty(Mono.error(Exception("No API available"))) } override fun getHead(): EthereumHead { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt new file mode 100644 index 00000000..a0646115 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -0,0 +1,102 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi +import org.reactivestreams.Subscriber +import reactor.core.publisher.EmitterProcessor +import reactor.core.publisher.Flux +import java.time.Duration +import kotlin.math.max +import kotlin.math.min +import kotlin.math.pow +import kotlin.math.roundToLong +import kotlin.random.Random + +class FilteredApis( + allUpstreams: List, + private val matcher: Selector.Matcher, + pos: Int, + private val repeatLimit: Long, + jitter: Int +): ApiSource { + + companion object { + private const val DEFAULT_DELAY_STEP = 100 + private const val MAX_WAIT_MILLIS = 5000L + } + + constructor(allUpstreams: List, + matcher: Selector.Matcher, + pos: Int): this(allUpstreams, matcher, pos, 10, 7) + + constructor(allUpstreams: List, + matcher: Selector.Matcher): this(allUpstreams, matcher, 0, 10, 10) + + private val delay: Int + private val upstreams: List + + private val control = EmitterProcessor.create(32, false) + + init { + delay = if (jitter > 0) { + Random.nextInt(DEFAULT_DELAY_STEP - jitter, DEFAULT_DELAY_STEP + jitter) + } else { + DEFAULT_DELAY_STEP + } + + upstreams = if (allUpstreams.size == 1 || pos == 0 || allUpstreams.isEmpty()) { + allUpstreams + } else { + val safePosition = pos % allUpstreams.size + allUpstreams.subList(safePosition, allUpstreams.size) + allUpstreams.subList(0, safePosition) + } + } + + fun waitDuration(rawn: Long): Duration { + val n = max(rawn, 1) + val time = min( + (n.toDouble().pow(2.0) * delay).roundToLong(), + MAX_WAIT_MILLIS + ) + return Duration.ofMillis(time) + } + + override fun subscribe(subscriber: Subscriber) { + val first = Flux.fromIterable(upstreams) + val retries = (1 until repeatLimit).map { r -> + Flux.fromIterable(upstreams).delaySubscription(waitDuration(r)) + }.let { Flux.concat(it) } + + Flux.concat(first, retries) + .filter(Upstream::isAvailable) + .filter(matcher::matches) + .flatMap { it.getApi(matcher) } + .zipWith(control).map { it.t1 } + .subscribe(subscriber) + } + + override fun resolve() { + control.onComplete() + } + + override fun request(tries: Int) { + //TODO check the buffer size before submitting + repeat(tries) { + control.onNext(true) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt deleted file mode 100644 index 8b6fd7a0..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt +++ /dev/null @@ -1,59 +0,0 @@ -/** - * Copyright (c) 2019 ETCDEV GmbH - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi - -class FilteringApiIterator( - private val upstreams: List, - private var pos: Int, - private val matcher: Selector.Matcher, - private val repeatLimit: Int = 5 -): Iterator { - - private var nextUpstream: Upstream? = null - private var consumed = 0 - - private fun nextInternal(): Boolean { - if (nextUpstream != null) { - return true - } - while (nextUpstream == null) { - consumed++ - if (consumed > upstreams.size * repeatLimit) { - return false - } - val upstream = upstreams[pos++ % upstreams.size] - if (upstream.isAvailable() && matcher.matches(upstream)) { - nextUpstream = upstream - } - } - return nextUpstream != null - } - - override fun hasNext(): Boolean { - return nextInternal() - } - - override fun next(): DirectEthereumApi { - if (nextInternal()) { - val curr = nextUpstream!! - nextUpstream = null - return curr.getApi(matcher) - } - throw IllegalStateException("No upstream API available") - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 0ef63858..8102494d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -19,13 +19,14 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import reactor.core.publisher.Flux +import reactor.core.publisher.Mono interface Upstream { fun isAvailable(): Boolean fun getStatus(): UpstreamAvailability fun observeStatus(): Flux fun getHead(): EthereumHead - fun getApi(matcher: Selector.Matcher): DirectEthereumApi + fun getApi(matcher: Selector.Matcher): Mono fun getOptions(): UpstreamsConfig.Options fun setLag(lag: Long) fun getLag(): Long diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt index e93196f3..a553f35b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.Commands import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import java.time.Duration import java.util.concurrent.TimeUnit @@ -28,28 +29,29 @@ class UpstreamValidator( private val options: UpstreamsConfig.Options ) { - fun validate(): UpstreamAvailability { + fun validate(): Mono { val batch = Batch() val peerCount = batch.add(Commands.net().peerCount()) val syncing = batch.add(Commands.eth().syncing()) - try { - ethereumUpstream.getApi(Selector.empty).rpcClient.execute(batch).get(5, TimeUnit.SECONDS) - if (syncing.get().isSyncing) { - return UpstreamAvailability.SYNCING - } - if (options.minPeers != null && peerCount.get() < options.minPeers!!) { - return UpstreamAvailability.IMMATURE - } - return UpstreamAvailability.OK - } catch (e: Throwable) { - return UpstreamAvailability.UNAVAILABLE - } + return ethereumUpstream.getApi(Selector.empty) + .map { api -> api.rpcClient.execute(batch) } + .flatMap { Mono.fromCompletionStage(it) } + .timeout(Duration.ofSeconds(10)) + .map { + if (syncing.get().isSyncing) { + UpstreamAvailability.SYNCING + } else if (options.minPeers != null && peerCount.get() < options.minPeers!!) { + UpstreamAvailability.IMMATURE + } else { + UpstreamAvailability.OK + } + }.onErrorContinue { _, _ -> UpstreamAvailability.UNAVAILABLE } } fun start(): Flux { return Flux.interval(Duration.ofSeconds(15)) - .map { + .flatMap { validate() - }.onErrorContinue { _, _ -> UpstreamAvailability.UNAVAILABLE } + } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index 6e2c5d64..5d132ec4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -21,6 +21,7 @@ import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable +import reactor.core.publisher.Mono import java.time.Duration open class EthereumUpstream( @@ -102,8 +103,8 @@ open class EthereumUpstream( return head } - override fun getApi(matcher: Selector.Matcher): DirectEthereumApi { - return api + override fun getApi(matcher: Selector.Matcher): Mono { + return Mono.just(api) } override fun getOptions(): UpstreamsConfig.Options { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt index ab29b001..f6c967cd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -121,7 +121,7 @@ open class GrpcUpstream( curr == null || curr.totalDifficulty < block.totalDifficulty }.flatMap { getApi(Selector.EmptyMatcher()) - .executeAndConvert(Commands.eth().getBlock(it.hash)) + .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(it.hash)) } .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream"))) .doOnError { t -> val msg = "Failed to download block data for chain $chain" @@ -194,8 +194,8 @@ open class GrpcUpstream( return head } - override fun getApi(matcher: Selector.Matcher): DirectEthereumApi { - return createApi(matcher) + override fun getApi(matcher: Selector.Matcher): Mono { + return Mono.just(createApi(matcher)) } override fun getOptions(): UpstreamsConfig.Options { diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index eed31de6..8ea5a2d9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -113,7 +113,7 @@ class NativeCallSpec extends Specification { nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) def delta = System.currentTimeMillis() - t1 then: - delta >= 200 + delta >= 100 } def "One call has no pause"() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy new file mode 100644 index 00000000..e1515c3a --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy @@ -0,0 +1,76 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.test + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.upstream.CallMethods +import io.emeraldpay.dshackle.upstream.DirectCallMethods +import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi +import io.infinitape.etherjar.rpc.Batch +import io.infinitape.etherjar.rpc.ExecutableBatch +import io.infinitape.etherjar.rpc.RpcCall +import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.transport.BatchStatus + +import java.util.concurrent.CompletableFuture + +class EthereumApiStub extends DirectEthereumApi { + + private String id + private static ObjectMapper objectMapper = TestingCommons.objectMapper() + private static RpcClient rpcClient = new RpcClientMock(); + + EthereumApiStub(Integer id) { + this(id.toString()) + } + + EthereumApiStub(String id) { + super(rpcClient, objectMapper, new DirectCallMethods()) + this.id = id + } + + @Override + String toString() { + return "API Stub $id" + } + + static class RpcClientMock implements RpcClient { + @Override + CompletableFuture execute(Batch batch) { + return null + } + + @Override + def CompletableFuture execute(RpcCall call) { + return null + } + + @Override + ExecutableBatch batch() { + return null + } + + @Override + EthCommands eth() { + return null + } + + @Override + TraceCommands trace() { + return null + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy new file mode 100644 index 00000000..d0ffbc58 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -0,0 +1,182 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.test.EthereumApiStub +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.DefaultRpcClient +import io.infinitape.etherjar.rpc.RpcClient +import reactor.test.StepVerifier +import spock.lang.Retry +import spock.lang.Specification + +import java.time.Duration + +class FilteredApisSpec extends Specification { + + def rpcClient = new DefaultRpcClient(null) + def objectMapper = TestingCommons.objectMapper() + def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM) + + def "Verifies labels"() { + setup: + List upstreams = [ + [test: "foo"], + [test: "bar"], + [test: "foo", test2: "baz"], + [test: "foo"], + [test: "baz"] + ].collect { + new EthereumUpstream( + "test", + Chain.ETHEREUM, + new DirectEthereumApi(rpcClient, objectMapper, ethereumTargets), + (EthereumWs) null, + new UpstreamsConfig.Options(), + new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap(it)), + ethereumTargets + ) + } + def matcher = new Selector.LabelMatcher("test", ["foo"]) + upstreams.forEach { + it.setLag(0) + it.setStatus(UpstreamAvailability.OK) + } + when: + def iter = new FilteredApis(upstreams, matcher, 0, 1, 0) + iter.request(10) + then: + StepVerifier.create(iter) + .expectNext(upstreams[0].api) + .expectNext(upstreams[2].api) + .expectNext(upstreams[3].api) + .expectComplete() + .verify(Duration.ofSeconds(1)) + + when: + iter = new FilteredApis(upstreams, matcher, 1, 1, 0) + iter.request(10) + then: + StepVerifier.create(iter) + .expectNext(upstreams[2].api) + .expectNext(upstreams[3].api) + .expectNext(upstreams[0].api) + .expectComplete() + .verify(Duration.ofSeconds(1)) + + when: + iter = new FilteredApis(upstreams, matcher, 1, 2, 0) + iter.request(10) + then: + StepVerifier.create(iter) + .expectNext(upstreams[2].api) + .expectNext(upstreams[3].api) + .expectNext(upstreams[0].api) + .expectNext(upstreams[2].api) + .expectNext(upstreams[3].api) + .expectNext(upstreams[0].api) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Exponential backoff"() { + setup: + def apis = new FilteredApis([], Selector.empty, 0, 1, 0) + expect: + wait == apis.waitDuration(n).toMillis() as Integer + where: + n | wait + 0 | 100 + 1 | 100 + 2 | 400 + 3 | 900 + 4 | 1600 + 5 | 2500 + 6 | 3600 + 7 | 4900 + 8 | 5000 + 9 | 5000 + 10 | 5000 + -1 | 100 + } + + @Retry + def "Backoff uses jitter"() { + setup: + def apis = new FilteredApis([], Selector.empty, 0, 1, 20) + when: + def act = apis.waitDuration(1).toMillis() + println act + then: + act >= 80 + act <= 120 + act != 100 + + when: + act = apis.waitDuration(3).toMillis() + println act + then: + act >= 900 - 9 * 20 + act <= 900 + 9 * 20 + act != 900 + } + + def "Makes pause between batches"() { + when: + def api1 = TestingCommons.api(Stub(RpcClient)) + def api2 = TestingCommons.api(Stub(RpcClient)) + def up1 = TestingCommons.upstream(api1) + def up2 = TestingCommons.upstream(api2) + then: + StepVerifier.withVirtualTime({ + def apis = new FilteredApis([up1, up2], Selector.empty, 0, 4, 0) + apis.request(10) + return apis + }) + .expectNext(api1, api2).as("Batch 1") + .expectNoEvent(Duration.ofMillis(100)).as("Wait 1") + .expectNext(api1, api2).as("Batch 2") + .expectNoEvent(Duration.ofMillis(400)).as("Wait 2") + .expectNext(api1, api2).as("Batch 3") + .expectNoEvent(Duration.ofMillis(900)).as("Wait 3") + .expectNext(api1, api2).as("Batch 4") + .expectComplete() + .verify(Duration.ofSeconds(10)) + } + + def "Starts with right position"() { + setup: + def apis = (0..5).collect { + new EthereumApiStub(it) + } + def ups = apis.collect { + TestingCommons.upstream(it) + } + when: + def act = new FilteredApis(ups, Selector.empty, 2, 1, 0) + act.request(10) + then: + StepVerifier.create(act) + .expectNext(apis[2], apis[3], apis[4], apis[5], apis[0], apis[1]) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy deleted file mode 100644 index 925abf7f..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy +++ /dev/null @@ -1,96 +0,0 @@ -/** - * Copyright (c) 2019 ETCDEV GmbH - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi -import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs -import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.DefaultRpcClient -import spock.lang.Specification - -class FilteringApiIteratorSpec extends Specification { - - def rpcClient = new DefaultRpcClient(null) - def objectMapper = TestingCommons.objectMapper() - def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM) - - def "Verifies labels"() { - setup: - List upstreams = [ - [test: "foo"], - [test: "bar"], - [test: "foo", test2: "baz"], - [test: "foo"], - [test: "baz"] - ].collect { - new EthereumUpstream( - "test", - Chain.ETHEREUM, - new DirectEthereumApi(rpcClient, objectMapper, ethereumTargets), - (EthereumWs) null, - new UpstreamsConfig.Options(), - new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap(it)), - ethereumTargets - ) - } - def matcher = new Selector.LabelMatcher("test", ["foo"]) - upstreams.forEach { - it.setLag(0) - it.setStatus(UpstreamAvailability.OK) - } - when: - def iter = new FilteringApiIterator(upstreams, 0, matcher, 1) - then: - iter.hasNext() - iter.next() == upstreams[0].api - iter.hasNext() - iter.next() == upstreams[2].api - iter.hasNext() - iter.next() == upstreams[3].api - !iter.hasNext() - - when: - iter = new FilteringApiIterator(upstreams, 1, matcher, 1) - then: - iter.hasNext() - iter.next() == upstreams[2].api - iter.hasNext() - iter.next() == upstreams[3].api - iter.hasNext() - iter.next() == upstreams[0].api - !iter.hasNext() - - when: - iter = new FilteringApiIterator(upstreams, 1, matcher, 2) - then: - iter.hasNext() - iter.next() == upstreams[2].api - iter.hasNext() - iter.next() == upstreams[3].api - iter.hasNext() - iter.next() == upstreams[0].api - iter.hasNext() - iter.next() == upstreams[2].api - iter.hasNext() - iter.next() == upstreams[3].api - iter.hasNext() - iter.next() == upstreams[0].api - !iter.hasNext() - } -}