diff --git a/docs/04-upstream-config.adoc b/docs/04-upstream-config.adoc index edbb206a..e672b61f 100644 --- a/docs/04-upstream-config.adoc +++ b/docs/04-upstream-config.adoc @@ -55,6 +55,7 @@ cluster: key: client.p8.key - id: infura-eth chain: ethereum + role: fallback labels: provider: infura options: @@ -80,23 +81,38 @@ There're two main segments for upstreams configuration: In the example above we have: -- default configuration for _Ethereum Mainnet_ which accepts upstream as valid when it not in fast synchronization mode -and has at least 10 peers.For _Kovan Testnet_ nodes the requirements are much relieved +- default configuration for _Ethereum Mainnet_ which accepts upstream as valid when it not in fast synchronization mode and has at least 10 peers.For _Kovan Testnet_ nodes the requirements are much relieved - as upstreams it has 2 configurations * balancer connects to another Dshackle/another machine by using gRPC protocol ** accepts (i.e. proxies) any blockchain available on that remote ** verifies TLS certificate of the server -** uses client certificate for authentication, i.e. remote server is accepting only clients authenticated by a -certificate +** uses client certificate for authentication, i.e. remote server is accepting only clients authenticated by a certificate * connects to Infura provided _Ethereum Mainnet_ -** configuration is using placeholders for `${INFURA_USER}` and `${INFURA_PASSWD}` which will be replaced with -corresponding environment variables values +** as a _fallback_ upstream, which means that it's used only if `us-nodes` fails +** configuration is using placeholders for `${INFURA_USER}` and `${INFURA_PASSWD}` which will be replaced with corresponding environment variables values ** uses Basic Authentication to authenticate requests on Infura -** label `[provider: infura]` is set for that particular upstream, which can be selected during a request.For example for -some requests you may want to use nodes with that label only, i.e. _"send that tx to infura nodes only"_, -or _"read only from archive node, with label [archive: true]"_ +** label `[provider: infura]` is set for that particular upstream, which can be selected during a request.For example for some requests you may want to use nodes with that label only, i.e. _"send that tx to infura nodes only"_, or _"read only from archive node, with label [archive: true]"_ ** upstream validation (peers, sync status, etc) is disabled for that particular upstream +=== Fallback upstream + +By default, the Dshackle connects to each upstream in a Round-Robin basis, i.e. sequentially one by one. +But if an upstream have `role: fallback` then it's used only in additional to other (_standard_) upstreams when their responses are not enough to finalize the request. + +Dshackle always starts with making requests to standard upstreams. +If all of them failed, if responses are inconsistent (ex. for `eth_getTransactionCount`), or when it needs to broadcast to wider networks (`sendrawtransaction`), then upstreams with role `fallback` are also used. + +The internal request order is: + +1. connect to each standard upstream +2. delay +3. try again to connect to standard upstreams +4. try to connect to fallback upstreams + +Steps 2-4 are repeated until a valid response received, or timeout for the original request is reached. + +In general, you set role `fallback` only to external nodes provided by a third party, when you want to use it as a last resort. + === Configuration options Options (default or as part of upstream config): @@ -106,7 +122,7 @@ Options (default or as part of upstream config): | Option | Default | Description | `disable-validation` | false | if `true` then Dshackle will not try to verify status of the upstream (could be useful for a trusted cloud - provider such as Infura, but disabling it is not recommended for a normal node) +provider such as Infura, but disabling it is not recommended for a normal node) | `min-peers` | 3 | specify minimum amount of connected peers, Dshackle will not use upstream with less than specified number | `timeout` | 60 | timeout in seconds after which request to the upstream will be discarded (and may be retried on an another upstream) |=== diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index 1147a8e5..0578b840 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -396,6 +396,7 @@ configuration, and may be omitted for most of the situations. ---- - id: local chain: ethereum + role: standard labels: fullnode: true methods: @@ -424,6 +425,11 @@ configuration, and may be omitted for most of the situations. | yes | Per-cluster identifier of an upstream +| `role` +| no +| `standard` (default) or `fallback`. +Fallback role mean that the upstream is used only after other upstreams failed or didn't return quorum + | `chain` | yes | Blockchain which is the provided by the upstream. diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index fe0ab0a5..da6f74d8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -74,6 +74,7 @@ class UpstreamsConfig { var connection: T? = null val labels = Labels() var methods: Methods? = null + var role: UpstreamRole = UpstreamRole.STANDARD @Suppress("unchecked") fun cast(type: Class): Upstream { @@ -84,6 +85,11 @@ class UpstreamsConfig { } } + enum class UpstreamRole { + STANDARD, + FALLBACK + } + open class UpstreamConnection open class RpcConnection : UpstreamConnection() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 3a5d6bf7..145d9bca 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -23,6 +23,7 @@ import org.yaml.snakeyaml.nodes.MappingNode import org.yaml.snakeyaml.nodes.ScalarNode import reactor.util.function.Tuples import java.io.InputStream +import java.lang.IllegalArgumentException import java.net.URI import java.time.Duration @@ -185,6 +186,15 @@ class UpstreamsConfigReader( internal fun readUpstreamStandard(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) { upstream.chain = getValueAsString(upNode, "chain") + getValueAsString(upNode, "role")?.let { + val name = it.trim() + try { + val role = UpstreamsConfig.UpstreamRole.valueOf(name.toUpperCase()) + upstream.role = role + } catch (e: IllegalArgumentException) { + log.warn("Unsupported role `$name` for upstream ${upstream.id}") + } + } if (hasAny(upNode, "labels")) { getMapping(upNode, "labels")?.let { labels -> labels.value.stream() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index b3cf793b..726eb779 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -145,7 +145,8 @@ open class ConfiguredUpstreams( val methods = buildMethods(config, chain) val upstream = BitcoinUpstream(config.id ?: "bitcoin-${seq.getAndIncrement()}", chain, directApi, - options, QuorumForLabels.QuorumItem(1, config.labels), + options, config.role, + QuorumForLabels.QuorumItem(1, config.labels), methods) upstream.start() @@ -183,7 +184,8 @@ open class ConfiguredUpstreams( log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") val ethereumUpstream = EthereumUpstream( config.id!!, - chain, directApi, wsFactoryApi, options, + chain, directApi, wsFactoryApi, + options, config.role, QuorumForLabels.QuorumItem(1, config.labels), methods ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 62f2b4ac..75a0b6ce 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -27,10 +27,12 @@ abstract class DefaultUpstream( defaultLag: Long, defaultAvail: UpstreamAvailability, private val options: UpstreamsConfig.Options, + private val role: UpstreamsConfig.UpstreamRole, private val targets: CallMethods? ) : Upstream { - constructor(id: String, options: UpstreamsConfig.Options, targets: CallMethods?) : this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, targets) + constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?) : + this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets) private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail))) private val statusStream: TopicProcessor = TopicProcessor.create() @@ -85,6 +87,10 @@ abstract class DefaultUpstream( return options } + override fun getRole(): UpstreamsConfig.UpstreamRole { + return role; + } + override fun getMethods(): CallMethods { return targets ?: throw IllegalStateException("Methods are not set") } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 93845663..d09c6434 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -16,9 +16,7 @@ */ package io.emeraldpay.dshackle.upstream -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.dshackle.config.UpstreamsConfig import org.reactivestreams.Subscriber import reactor.core.publisher.EmitterProcessor import reactor.core.publisher.Flux @@ -33,13 +31,26 @@ class FilteredApis( allUpstreams: List, private val matcher: Selector.Matcher, pos: Int, - private val repeatLimit: Long, + /** + * Limit of retries + */ + private val retryLimit: Long, jitter: Int ) : ApiSource { companion object { private const val DEFAULT_DELAY_STEP = 100 private const val MAX_WAIT_MILLIS = 5000L + + @JvmStatic + fun startFrom(upstreams: List, pos: Int): List { + return if (upstreams.size <= 1 || pos == 0) { + upstreams + } else { + val safePosition = pos % upstreams.size + upstreams.subList(safePosition, upstreams.size) + upstreams.subList(0, safePosition) + } + } } constructor(allUpstreams: List, @@ -50,7 +61,8 @@ class FilteredApis( matcher: Selector.Matcher) : this(allUpstreams, matcher, 0, 10, 10) private val delay: Int - private val upstreams: List + private val standardUpstreams: List + private val standardWithFallback: List private val control = EmitterProcessor.create(32, false) @@ -61,12 +73,20 @@ class FilteredApis( 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) + standardUpstreams = allUpstreams.filter { + it.getRole() == UpstreamsConfig.UpstreamRole.STANDARD + }.let { + startFrom(it, pos) } + val fallbackUpstreams = allUpstreams.filter { + it.getRole() == UpstreamsConfig.UpstreamRole.FALLBACK + }.let { + startFrom(it, pos) + } + standardWithFallback = emptyList() + .plus(standardUpstreams) + .plus(fallbackUpstreams) + } fun waitDuration(rawn: Long): Duration { @@ -79,9 +99,15 @@ class FilteredApis( } override fun subscribe(subscriber: Subscriber) { - val first = Flux.fromIterable(upstreams) - val retries = (1 until repeatLimit).map { r -> - Flux.fromIterable(upstreams).delaySubscription(waitDuration(r)) + // initially try only standard upstreams + val first = Flux.fromIterable(standardUpstreams) + // if all failed, try both standard and fallback upstreams, repeating in cycle + val retries = (0 until (retryLimit - 1)).map { r -> + Flux.fromIterable(standardWithFallback) + // add delay to let upstream to restore if it's a temp failure + // but delay only start of the check, not between upstreams + // i.e. if all upstreams failed -> wait -> check all without waiting in between + .delaySubscription(waitDuration(r + 1)) }.let { Flux.concat(it) } Flux.concat(first, retries) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 3e168d2b..7b126d01 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -145,10 +145,16 @@ abstract class Multistream( else upstreams.map { it.getStatus() }.min()!! } + //TODO options for multistream are useless override fun getOptions(): UpstreamsConfig.Options { return UpstreamsConfig.Options() } + //TODO roles for multistream are useless + override fun getRole(): UpstreamsConfig.UpstreamRole { + return UpstreamsConfig.UpstreamRole.STANDARD + } + override fun getMethods(): CallMethods { return callMethods ?: throw IllegalStateException("Methods are not initialized yet") } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 3af96e79..0dce5914 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -31,6 +31,7 @@ interface Upstream { fun getHead(): Head fun getApi(): Reader fun getOptions(): UpstreamsConfig.Options + fun getRole(): UpstreamsConfig.UpstreamRole fun setLag(lag: Long) fun getLag(): Long fun getLabels(): Collection diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt index b657a0c6..fce79cf2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt @@ -34,9 +34,10 @@ open class BitcoinUpstream( val chain: Chain, private val directApi: Reader, options: UpstreamsConfig.Options, + role: UpstreamsConfig.UpstreamRole, val node: QuorumForLabels.QuorumItem, callMethods: CallMethods -) : DefaultUpstream(id, options, callMethods), Lifecycle { +) : DefaultUpstream(id, options, role, callMethods), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java) 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 f5b9df51..b1d5e484 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -40,13 +40,16 @@ open class EthereumUpstream( private val directReader: Reader, private val ethereumWsFactory: EthereumWsFactory? = null, options: UpstreamsConfig.Options, + role: UpstreamsConfig.UpstreamRole, val node: QuorumForLabels.QuorumItem, targets: CallMethods -) : DefaultUpstream(id, options, targets), Upstream, CachesEnabled, Lifecycle { +) : DefaultUpstream(id, options, role, targets), Upstream, CachesEnabled, Lifecycle { - constructor(id: String, chain: Chain, api: Reader) : this(id, chain, api, null, - UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), - DirectCallMethods()) + constructor(id: String, chain: Chain, api: Reader) : + this(id, chain, api, null, + UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD, + QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), + DirectCallMethods()) private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index d473b4ce..7288f1bb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -60,6 +60,7 @@ open class EthereumGrpcUpstream( ) : DefaultUpstream( "$parentId/${chain.chainCode}", UpstreamsConfig.Options.getDefaults(), + UpstreamsConfig.UpstreamRole.STANDARD, null ), Lifecycle { diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 5c9d9059..f2ed56ee 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -202,4 +202,40 @@ class UpstreamsConfigReaderSpec extends Specification { where: id << ["test", "test_test", "test-test", "test123", "test1test", "foo_bar_12"] } + + def "Parse config without fallback role"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-basic.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 2 + act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD + act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.STANDARD + } + + def "Parse config with fallback role"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-roles.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 2 + act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD + act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.FALLBACK + } + + def "Parse config with invalid role"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-roles-invalid.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 2 + act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD + act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.STANDARD + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index 61149bca..91939c8a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.quorum +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.FilteredApis @@ -34,6 +35,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD 1 * getApi() >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(JsonRpcResponse.ok("1")) } @@ -60,6 +62,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD _ * getApi() >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.error(1, "test")), @@ -89,6 +92,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD _ * getApi() >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.ok("null")), @@ -119,6 +123,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD _ * getApi() >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.error(1, "test")), @@ -148,6 +153,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD _ * getApi() >> Mock(Reader) { 3 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.ok("null")), @@ -178,6 +184,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD _ * getApi() >> Mock(Reader) { 3 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.ok("null")), diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index c42e54e6..5b10b7c6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -61,7 +61,9 @@ class EthereumUpstreamMock extends EthereumUpstream { EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods) { super(id, chain, api, null, - UpstreamsConfig.Options.getDefaults(), new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), + UpstreamsConfig.Options.getDefaults(), + UpstreamsConfig.UpstreamRole.STANDARD, + new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), methods) setLag(0) setStatus(UpstreamAvailability.OK) @@ -85,4 +87,9 @@ class EthereumUpstreamMock extends EthereumUpstream { Head getHead() { return ethereumHeadMock } + + @Override + String toString() { + return "Upstream mock ${getId()}" + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index d51d9b19..22158104 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -43,6 +43,10 @@ class TestingCommons { return new EthereumApiMock() } + static EthereumUpstreamMock upstream(String id, Reader api) { + return new EthereumUpstreamMock(id, Chain.ETHEREUM, api) + } + static EthereumUpstreamMock upstream(Reader api) { return new EthereumUpstreamMock(Chain.ETHEREUM, api) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 5f2d06e6..88447fb0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory import io.emeraldpay.grpc.Chain +import reactor.core.publisher.Flux import reactor.test.StepVerifier import spock.lang.Retry import spock.lang.Specification @@ -50,6 +51,7 @@ class FilteredApisSpec extends Specification { TestingCommons.api().tap { it.id = "${i++}" }, (EthereumWsFactory) null, new UpstreamsConfig.Options(), + UpstreamsConfig.UpstreamRole.STANDARD, new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)), ethereumTargets ) @@ -175,7 +177,52 @@ class FilteredApisSpec extends Specification { then: StepVerifier.create(act) .expectNext(ups[2], ups[3], ups[4], ups[5], ups[0], ups[1]) - .expectComplete() - .verify(Duration.ofSeconds(1)) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Start with offset"() { + expect: + FilteredApis.startFrom([0, 1, 2, 3, 4], pos) == exp + where: + pos | exp + 0 | [0, 1, 2, 3, 4] + 1 | [1, 2, 3, 4, 0] + 2 | [2, 3, 4, 0, 1] + 3 | [3, 4, 0, 1, 2] + 4 | [4, 0, 1, 2, 3] + 5 | [0, 1, 2, 3, 4] + 6 | [1, 2, 3, 4, 0] + } + + def "Starts with standard"() { + setup: + List standard = (0..1).collect { + TestingCommons.upstream( + it.toString(), + new EthereumApiStub(it) + ) + } + def fallback = [ + Mock(Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK + _ * isAvailable() >> true + } + ] + when: + def act = new FilteredApis([] + fallback + standard, + Selector.empty, 0, 3, 0) + act.request(10) + then: + StepVerifier.create(act) + .expectNext(standard[0], standard[1]).as("Initial requests") + + .expectNext(standard[0], standard[1]).as("Retry with standard") + .expectNext(fallback[0]).as("Retry with fallback") + + .expectNext(standard[0], standard[1]).as("Second retry with standard") + .expectNext(fallback[0]).as("Second retry with fallback") + .expectComplete() + .verify(Duration.ofSeconds(1)) } } diff --git a/src/test/resources/upstreams-roles-invalid.yaml b/src/test/resources/upstreams-roles-invalid.yaml new file mode 100644 index 00000000..da9c1f1c --- /dev/null +++ b/src/test/resources/upstreams-roles-invalid.yaml @@ -0,0 +1,31 @@ +version: v1 + +defaults: + - chains: + - ethereum + options: + min-peers: 3 + +upstreams: + - id: local + chain: ethereum + connection: + ethereum: + rpc: + url: "http://localhost:8545" + ws: + url: "ws://localhost:8546" + origin: "http://localhost" + basic-auth: + username: 9c199ad8f281f20154fc258fe41a6814 + password: 258fe4149c199ad8f2811a68f20154fc + - id: infura + chain: ethereum + role: fallbackZ + connection: + ethereum: + rpc: + url: "https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2" + basic-auth: + username: 4fc258fe41a68149c199ad8f281f2015 + password: 1a68f20154fc258fe4149c199ad8f281 \ No newline at end of file diff --git a/src/test/resources/upstreams-roles.yaml b/src/test/resources/upstreams-roles.yaml new file mode 100644 index 00000000..2409d7f0 --- /dev/null +++ b/src/test/resources/upstreams-roles.yaml @@ -0,0 +1,31 @@ +version: v1 + +defaults: + - chains: + - ethereum + options: + min-peers: 3 + +upstreams: + - id: local + chain: ethereum + connection: + ethereum: + rpc: + url: "http://localhost:8545" + ws: + url: "ws://localhost:8546" + origin: "http://localhost" + basic-auth: + username: 9c199ad8f281f20154fc258fe41a6814 + password: 258fe4149c199ad8f2811a68f20154fc + - id: infura + chain: ethereum + role: fallback + connection: + ethereum: + rpc: + url: "https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2" + basic-auth: + username: 4fc258fe41a68149c199ad8f281f2015 + password: 1a68f20154fc258fe4149c199ad8f281 \ No newline at end of file