problem: suboptimal routing for upstreams in different datacenters

solution: additional role to separate primary and secondary upstreams
fix: #141

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 <Z : UpstreamConnection> cast(type: Class<Z>): Upstream<Z> {
@@ -87,7 +87,8 @@ open class UpstreamsConfig {
     }
 
     enum class UpstreamRole {
-        STANDARD,
+        PRIMARY,
+        SECONDARY,
         FALLBACK
     }
 

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

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,

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<Upstream>
+    private val primaryUpstreams: List<Upstream>
+    private val secondaryUpstreams: List<Upstream>
     private val standardWithFallback: List<Upstream>
 
     private val control = Sinks.many().unicast().onBackpressureBuffer<Boolean>()
@@ -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<Upstream>()
-            .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<in Upstream>) {
         // 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")

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 {

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

@@ -33,7 +33,7 @@ open class EthereumRpcUpstream(
     constructor(id: String, chain: Chain, api: Reader<JsonRpcRequest, JsonRpcResponse>) :
         this(
             id, chain, api, null,
-            UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD,
+            UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.PRIMARY,
             QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
             DirectCallMethods()
         )

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 {

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,

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()