From c060aeda311c7d162c9f093824edb43d4bb3ba9b Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 5 Jan 2022 18:42:35 -0500 Subject: [PATCH] problem: suboptimal routing for upstreams in different datacenters solution: additional role to separate primary and secondary upstreams fix: #141 --- docs/04-upstream-config.adoc | 33 +++++++++----- .../dshackle/config/UpstreamsConfig.kt | 5 ++- .../dshackle/config/UpstreamsConfigReader.kt | 5 ++- .../dshackle/startup/ConfiguredUpstreams.kt | 1 + .../dshackle/upstream/FilteredApis.kt | 33 +++++++++----- .../dshackle/upstream/Multistream.kt | 2 +- .../upstream/ethereum/EthereumRpcUpstream.kt | 2 +- .../upstream/grpc/BitcoinGrpcUpstream.kt | 3 +- .../upstream/grpc/EthereumGrpcUpstream.kt | 3 +- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 6 ++- .../config/UpstreamsConfigReaderSpec.groovy | 23 +++++++--- .../quorum/QuorumRpcReaderSpec.groovy | 14 +++--- .../dshackle/test/EthereumUpstreamMock.groovy | 2 +- .../dshackle/upstream/FilteredApisSpec.groovy | 45 ++++++++++++++++++- .../grpc/EthereumGrpcUpstreamSpec.groovy | 7 +-- src/test/resources/upstreams-roles-2.yaml | 44 ++++++++++++++++++ 16 files changed, 179 insertions(+), 49 deletions(-) create mode 100644 src/test/resources/upstreams-roles-2.yaml diff --git a/docs/04-upstream-config.adoc b/docs/04-upstream-config.adoc index 8eba3782..ebe27fb0 100644 --- a/docs/04-upstream-config.adoc +++ b/docs/04-upstream-config.adoc @@ -74,7 +74,7 @@ cluster: password: ${INFURA_PASSWD} ---- -There're two main segments for upstreams configuration: +There are two main segments for upstreams configuration: - _upstreams_ - a list of API to connect to, with all configuration specific for upstream and chain - and _default options_ as common configuration options applied to all nodes in that group @@ -94,24 +94,33 @@ In the example above we have: ** 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 +=== Roles and 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. +If you need more gradual control over the order of which upstream is used and when you can assign following roles: -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. +- `primary` (default role if nothing specified) +- `secondary` +- `fallback` -The internal request order is: +Where `primary` and `secondary` are considered here a _standard_ upstreams, and `fallback` is used on failure of standard upstreams. +I.e. the 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 a wider network (`sendrawtransaction`), then upstreams with role `fallback` cames to use. -1. connect to each standard upstream -2. delay -3. try again to connect to standard upstreams -4. try to connect to fallback upstreams +The internal request order is (goes to next only if all upstreams on current step a not available or failed): -Steps 2-4 are repeated until a valid response received, or timeout for the original request is reached. +1. tries with primary upstreams +2. tries with secondary upstream +3. ... delay (100ms at first, increased each iteration) +4. tries with primary upstreams +5. tries with secondary upstream +6. tries with fallback upstreams +7. ... go to step 3 -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. +Steps 3-6 are repeated until a valid response received, or a timeout for the original request is reached. + +In general: +- you set role `secondary` for upstream in another cluster/datacenter - you set role `fallback` for an external upstream which may be provided by a third party, and you want to use it as a last resort === Configuration options diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 09f4fae4..552a6ddf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -75,7 +75,7 @@ open class UpstreamsConfig { var connection: T? = null val labels = Labels() var methods: Methods? = null - var role: UpstreamRole = UpstreamRole.STANDARD + var role: UpstreamRole = UpstreamRole.PRIMARY @Suppress("UNCHECKED_CAST") fun cast(type: Class): Upstream { @@ -87,7 +87,8 @@ open class UpstreamsConfig { } enum class UpstreamRole { - STANDARD, + PRIMARY, + SECONDARY, FALLBACK } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index c477fd66..4bb2fae9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -214,7 +214,10 @@ class UpstreamsConfigReader( internal fun readUpstreamStandard(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) { upstream.chain = getValueAsString(upNode, "chain") getValueAsString(upNode, "role")?.let { - val name = it.trim() + val name = it.trim().let { + // `standard` was initial role, now split into `primary` and `secondary` + if (it == "standard") "primary" else it + } try { val role = UpstreamsConfig.UpstreamRole.valueOf(name.uppercase(Locale.getDefault())) upstream.role = role diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 489a79da..7e7ec7fa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -234,6 +234,7 @@ open class ConfiguredUpstreams( val endpoint = config.connection!! val ds = GrpcUpstreams( config.id!!, + config.role, endpoint.host!!, endpoint.port, endpoint.auth, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index ce015728..4a3519e6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -82,7 +82,8 @@ class FilteredApis( ) : this(chain, allUpstreams, matcher, 0, 10, 10) private val delay: Int - private val standardUpstreams: List + private val primaryUpstreams: List + private val secondaryUpstreams: List private val standardWithFallback: List private val control = Sinks.many().unicast().onBackpressureBuffer() @@ -94,8 +95,13 @@ class FilteredApis( DEFAULT_DELAY_STEP } - standardUpstreams = allUpstreams.filter { - it.getRole() == UpstreamsConfig.UpstreamRole.STANDARD + primaryUpstreams = allUpstreams.filter { + it.getRole() == UpstreamsConfig.UpstreamRole.PRIMARY + }.let { + startFrom(it, pos) + } + secondaryUpstreams = allUpstreams.filter { + it.getRole() == UpstreamsConfig.UpstreamRole.SECONDARY }.let { startFrom(it, pos) } @@ -105,12 +111,14 @@ class FilteredApis( startFrom(it, pos) } standardWithFallback = emptyList() - .plus(standardUpstreams) + .plus(primaryUpstreams) + .plus(secondaryUpstreams) .plus(fallbackUpstreams) if (Global.metricsExtended) { getMetrics(chain).let { monitoring -> - monitoring.countStd.record(standardUpstreams.size.toDouble()) + monitoring.countPrimary.record(primaryUpstreams.size.toDouble()) + monitoring.countSecondary.record(secondaryUpstreams.size.toDouble()) monitoring.countFallback.record(fallbackUpstreams.size.toDouble()) } } @@ -145,17 +153,18 @@ class FilteredApis( override fun subscribe(subscriber: Subscriber) { // initially try only standard upstreams - val first = Flux.fromIterable(standardUpstreams) + val first = Flux.fromIterable(primaryUpstreams) + val second = Flux.fromIterable(secondaryUpstreams) // 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 + // add a 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) } - var result = Flux.concat(first, retries) + var result = Flux.concat(first, second, retries) if (Global.metricsExtended) { var count = 0 @@ -186,9 +195,13 @@ class FilteredApis( } class Monitoring(chain: Chain) { - val countStd: DistributionSummary = DistributionSummary.builder("$metricsCode.exist") + val countPrimary: DistributionSummary = DistributionSummary.builder("$metricsCode.exist") .description("Count of available upstreams to select") - .tags(listOf(Tag.of("chain", chain.chainCode), Tag.of("role", "std"))) + .tags(listOf(Tag.of("chain", chain.chainCode), Tag.of("role", "primary"))) + .register(Metrics.globalRegistry) + val countSecondary: DistributionSummary = DistributionSummary.builder("$metricsCode.exist") + .description("Count of available upstreams to select") + .tags(listOf(Tag.of("chain", chain.chainCode), Tag.of("role", "secondary"))) .register(Metrics.globalRegistry) val countFallback: DistributionSummary = DistributionSummary.builder("$metricsCode.exist") .description("Count of available fallback upstreams to select") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 61802be7..6319cede 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -212,7 +212,7 @@ abstract class Multistream( // TODO roles for multistream are useless override fun getRole(): UpstreamsConfig.UpstreamRole { - return UpstreamsConfig.UpstreamRole.STANDARD + return UpstreamsConfig.UpstreamRole.PRIMARY } override fun getMethods(): CallMethods { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 6846da0b..51bed8d7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -33,7 +33,7 @@ open class EthereumRpcUpstream( constructor(id: String, chain: Chain, api: Reader) : this( id, chain, api, null, - UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD, + UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.PRIMARY, QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), DirectCallMethods() ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 8009da35..ad88eeb4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -45,6 +45,7 @@ import java.util.function.Function class BitcoinGrpcUpstream( private val parentId: String, + role: UpstreamsConfig.UpstreamRole, chain: Chain, val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, private val client: JsonRpcGrpcClient @@ -52,7 +53,7 @@ class BitcoinGrpcUpstream( "$parentId/${chain.chainCode}", chain, UpstreamsConfig.Options.getDefaults(), - UpstreamsConfig.UpstreamRole.STANDARD + role ), GrpcUpstream, Lifecycle { 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 e17bae7c..2ccebfd7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -48,13 +48,14 @@ import java.util.function.Function open class EthereumGrpcUpstream( private val parentId: String, + role: UpstreamsConfig.UpstreamRole, private val chain: Chain, private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, private val client: JsonRpcGrpcClient ) : EthereumUpstream( "$parentId/${chain.chainCode}", UpstreamsConfig.Options.getDefaults(), - UpstreamsConfig.UpstreamRole.STANDARD, + role, null, null ), GrpcUpstream, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 3d294efa..a2943e65 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -21,6 +21,7 @@ import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.config.AuthConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -51,6 +52,7 @@ import kotlin.concurrent.withLock class GrpcUpstreams( private val id: String, + private val role: UpstreamsConfig.UpstreamRole, private val host: String, private val port: Int, private val auth: AuthConfig.ClientTlsAuth? = null, @@ -198,7 +200,7 @@ class GrpcUpstreams( val current = known[chain] return if (current == null) { val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = EthereumGrpcUpstream(id, chain, client!!, rpcClient) + val created = EthereumGrpcUpstream(id, role, chain, client!!, rpcClient) created.timeout = this.timeout known[chain] = created created.start() @@ -214,7 +216,7 @@ class GrpcUpstreams( val current = known[chain] return if (current == null) { val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = BitcoinGrpcUpstream(id, chain, client!!, rpcClient) + val created = BitcoinGrpcUpstream(id, role, chain, client!!, rpcClient) created.timeout = this.timeout known[chain] = created created.start() diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 61bf80f9..73f8db20 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -343,8 +343,8 @@ class UpstreamsConfigReaderSpec extends Specification { then: act != null act.upstreams.size() == 2 - act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD - act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.STANDARD + act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.PRIMARY + act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.PRIMARY } def "Parse config with fallback role"() { @@ -355,10 +355,23 @@ class UpstreamsConfigReaderSpec extends Specification { then: act != null act.upstreams.size() == 2 - act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD + act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.PRIMARY act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.FALLBACK } + def "Parse config with secondary role"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-roles-2.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 3 + act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.PRIMARY + act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.SECONDARY + act.upstreams.get(2).role == UpstreamsConfig.UpstreamRole.FALLBACK + } + def "Parse config with invalid role"() { setup: def config = this.class.getClassLoader().getResourceAsStream("upstreams-roles-invalid.yaml") @@ -367,7 +380,7 @@ class UpstreamsConfigReaderSpec extends Specification { then: act != null act.upstreams.size() == 2 - act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD - act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.STANDARD + act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.PRIMARY + act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.PRIMARY } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index ab27e7ec..dbdc30de 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -38,7 +38,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true - _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY 1 * getApi() >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(JsonRpcResponse.ok("1")) } @@ -71,7 +71,7 @@ class QuorumRpcReaderSpec extends Specification { } def up = Mock(Upstream) { _ * isAvailable() >> true - _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getApi() >> api } def apis = new FilteredApis( @@ -98,7 +98,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true - _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getApi() >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.ok("null")), @@ -130,7 +130,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true - _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getApi() >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.error(1, "test")), @@ -161,7 +161,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true - _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getApi() >> Mock(Reader) { 3 * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.ok("null")), @@ -200,7 +200,7 @@ class QuorumRpcReaderSpec extends Specification { } def up = Mock(Upstream) { _ * isAvailable() >> true - _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getApi() >> api } def apis = new FilteredApis( @@ -226,7 +226,7 @@ class QuorumRpcReaderSpec extends Specification { def up = Mock(Upstream) { _ * getLag() >> 0 _ * isAvailable() >> true - _ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getApi() >> Mock(Reader) { _ * read(new JsonRpcRequest("eth_test", [])) >>> [ Mono.just(JsonRpcResponse.error(-3010, "test")), diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index 4e136ec4..4b292dd7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -63,7 +63,7 @@ class EthereumUpstreamMock extends EthereumRpcUpstream { EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods) { super(id, chain, api, null, UpstreamsConfig.Options.getDefaults(), - UpstreamsConfig.UpstreamRole.STANDARD, + UpstreamsConfig.UpstreamRole.PRIMARY, new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), methods) setLag(0) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 95b531c9..8ea9811e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -52,7 +52,7 @@ class FilteredApisSpec extends Specification { TestingCommons.api().tap { it.id = "${i++}" }, (EthereumWsFactory) null, new UpstreamsConfig.Options(), - UpstreamsConfig.UpstreamRole.STANDARD, + UpstreamsConfig.UpstreamRole.PRIMARY, new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)), ethereumTargets ) @@ -210,7 +210,7 @@ class FilteredApisSpec extends Specification { 6 | [0, 1] } - def "Starts with standard"() { + def "Starts with primary"() { setup: List standard = (0..1).collect { TestingCommons.upstream( @@ -241,4 +241,45 @@ class FilteredApisSpec extends Specification { .expectComplete() .verify(Duration.ofSeconds(1)) } + + def "Use secondary after primary"() { + setup: + List standard = (0..1).collect { + TestingCommons.upstream( + it.toString(), + new EthereumApiStub(it) + ) + } + List fallback = [ + Mock([name: "fallback"], Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK + _ * isAvailable() >> true + } + ] + List secondary = [ + Mock([name: "secondary"], Upstream) { + _ * getRole() >> UpstreamsConfig.UpstreamRole.SECONDARY + _ * isAvailable() >> true + } + ] + when: + def act = new FilteredApis(Chain.ETHEREUM, + [] + fallback + standard + secondary, + Selector.empty, 0, 3, 0) + act.request(11) + then: + StepVerifier.create(act) + .expectNext(standard[0], standard[1]).as("Initial requests with primary") + .expectNext(secondary[0]).as("Initial requests with secondary") + + .expectNext(standard[0], standard[1]).as("Retry with primary") + .expectNext(secondary[0]).as("Retry with secondary") + .expectNext(fallback[0]).as("Retry with fallback") + + .expectNext(standard[0], standard[1]).as("Second retry with primary") + .expectNext(secondary[0]).as("Second retry with secondary") + .expectNext(fallback[0]).as("Second retry with fallback") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } } 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 77a5e808..b98904a1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -22,6 +22,7 @@ import io.emeraldpay.api.proto.BlockchainGrpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.test.MockGrpcServer import io.emeraldpay.dshackle.test.TestingCommons @@ -80,7 +81,7 @@ class EthereumGrpcUpstreamSpec extends Specification { ) } }) - def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain, metrics)) + def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics)) upstream.setLag(0) upstream.update(BlockchainOuterClass.DescribeChain.newBuilder() .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) @@ -138,7 +139,7 @@ class EthereumGrpcUpstreamSpec extends Specification { ) } }) - def upstream = new EthereumGrpcUpstream("test", Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics)) + def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics)) upstream.setLag(0) upstream.update(BlockchainOuterClass.DescribeChain.newBuilder() .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) @@ -200,7 +201,7 @@ class EthereumGrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain, metrics)) + def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics)) upstream.setLag(0) upstream.update(BlockchainOuterClass.DescribeChain.newBuilder() .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) diff --git a/src/test/resources/upstreams-roles-2.yaml b/src/test/resources/upstreams-roles-2.yaml new file mode 100644 index 00000000..a74c317c --- /dev/null +++ b/src/test/resources/upstreams-roles-2.yaml @@ -0,0 +1,44 @@ +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: closeby + chain: ethereum + role: secondary + 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