Merge pull request #30 from emeraldpay/feat/fallback-role

solution: fallback upstreams

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
+- 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
-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
+** uses client certificate for authentication, i.e. remote server is accepting only clients authenticated by a certificate
-certificate
 * connects to Infura provided _Ethereum Mainnet_
-** configuration is using placeholders for `${INFURA_USER}` and `${INFURA_PASSWD}` which will be replaced with
+** as a _fallback_ upstream, which means that it's used only if `us-nodes` fails
-corresponding environment variables values
+** 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
+** 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]"_
-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)
 |===

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.

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 <Z : UpstreamConnection> cast(type: Class<Z>): Upstream<Z> {
@@ -84,6 +85,11 @@ class UpstreamsConfig {
         }
     }
 
+    enum class UpstreamRole {
+        STANDARD,
+        FALLBACK
+    }
+
     open class UpstreamConnection
 
     open class RpcConnection : UpstreamConnection() {

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

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
         )

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<UpstreamAvailability> = 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")
     }

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.config.UpstreamsConfig
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.reactivestreams.Subscriber
 import reactor.core.publisher.EmitterProcessor
 import reactor.core.publisher.Flux
@@ -33,13 +31,26 @@ class FilteredApis(
         allUpstreams: List<Upstream>,
         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 <T> startFrom(upstreams: List<T>, pos: Int): List<T> {
+            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<Upstream>,
@@ -50,7 +61,8 @@ class FilteredApis(
                 matcher: Selector.Matcher) : this(allUpstreams, matcher, 0, 10, 10)
 
     private val delay: Int
-    private val upstreams: List<Upstream>
+    private val standardUpstreams: List<Upstream>
+    private val standardWithFallback: List<Upstream>
 
     private val control = EmitterProcessor.create<Boolean>(32, false)
 
@@ -61,12 +73,20 @@ class FilteredApis(
             DEFAULT_DELAY_STEP
         }
 
-        upstreams = if (allUpstreams.size == 1 || pos == 0 || allUpstreams.isEmpty()) {
+        standardUpstreams = allUpstreams.filter {
-            allUpstreams
+            it.getRole() == UpstreamsConfig.UpstreamRole.STANDARD
-        } else {
+        }.let {
-            val safePosition = pos % allUpstreams.size
+            startFrom(it, pos)
-            allUpstreams.subList(safePosition, allUpstreams.size) + allUpstreams.subList(0, safePosition)
         }
+        val fallbackUpstreams = allUpstreams.filter {
+            it.getRole() == UpstreamsConfig.UpstreamRole.FALLBACK
+        }.let {
+            startFrom(it, pos)
+        }
+        standardWithFallback = emptyList<Upstream>()
+                .plus(standardUpstreams)
+                .plus(fallbackUpstreams)
+
     }
 
     fun waitDuration(rawn: Long): Duration {
@@ -79,9 +99,15 @@ class FilteredApis(
     }
 
     override fun subscribe(subscriber: Subscriber<in Upstream>) {
-        val first = Flux.fromIterable(upstreams)
+        // initially try only standard upstreams
-        val retries = (1 until repeatLimit).map { r ->
+        val first = Flux.fromIterable(standardUpstreams)
-            Flux.fromIterable(upstreams).delaySubscription(waitDuration(r))
+        // 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)

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")
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt

@@ -31,6 +31,7 @@ interface Upstream {
     fun getHead(): Head
     fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
     fun getOptions(): UpstreamsConfig.Options
+    fun getRole(): UpstreamsConfig.UpstreamRole
     fun setLag(lag: Long)
     fun getLag(): Long
     fun getLabels(): Collection<UpstreamsConfig.Labels>

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt

@@ -34,9 +34,10 @@ open class BitcoinUpstream(
         val chain: Chain,
         private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
         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)

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

@@ -40,13 +40,16 @@ open class EthereumUpstream(
         private val directReader: Reader<JsonRpcRequest, JsonRpcResponse>,
         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<JsonRpcRequest, JsonRpcResponse>) : this(id, chain, api, null,
+    constructor(id: String, chain: Chain, api: Reader<JsonRpcRequest, JsonRpcResponse>) :
-            UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
+            this(id, chain, api, null,
-            DirectCallMethods())
+                    UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD,
+                    QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
+                    DirectCallMethods())
 
 
     private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)

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 {
 

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
+    }
 }

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")),

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<JsonRpcRequest, JsonRpcResponse> 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()}"
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy

@@ -43,6 +43,10 @@ class TestingCommons {
         return new EthereumApiMock()
     }
 
+    static EthereumUpstreamMock upstream(String id, Reader<JsonRpcRequest, JsonRpcResponse> api) {
+        return new EthereumUpstreamMock(id, Chain.ETHEREUM, api)
+    }
+
     static EthereumUpstreamMock upstream(Reader<JsonRpcRequest, JsonRpcResponse> api) {
         return new EthereumUpstreamMock(Chain.ETHEREUM, api)
     }

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()
+                .expectComplete()
-            .verify(Duration.ofSeconds(1))
+                .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<Upstream> 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))
     }
 }

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

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