Merge pull request #138 from p2p-org/make-more-soft-availability

Adjustments in availability determination:

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

@@ -65,7 +65,7 @@ abstract class DefaultUpstream(
     }
 
     override fun isAvailable(): Boolean {
-        return getStatus() == UpstreamAvailability.OK
+        return getStatus() == UpstreamAvailability.OK || getStatus() == UpstreamAvailability.LAGGING
     }
 
     fun onStatus(value: BlockchainOuterClass.ChainStatus) {

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

@@ -160,11 +160,11 @@ class FilteredApis(
 
     override fun subscribe(subscriber: Subscriber<in Upstream>) {
         // initially try only standard upstreams
-        val first = Flux.fromIterable(primaryUpstreams)
+        val first = Flux.fromIterable(primaryUpstreams.sortedBy { it.getStatus().grpcId })
-        val second = Flux.fromIterable(secondaryUpstreams)
+        val second = Flux.fromIterable(secondaryUpstreams.sortedBy { it.getStatus().grpcId })
         // if all failed, try both standard and fallback upstreams, repeating in cycle
         val retries = (0 until (retryLimit - 1)).map { r ->
-            Flux.fromIterable(standardWithFallback)
+            Flux.fromIterable(standardWithFallback.sortedBy { it.getStatus().grpcId })
                 // 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

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy

@@ -227,6 +227,7 @@ class FilteredApisSpec extends Specification {
                 Mock(Upstream) {
                     _ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK
                     _ * isAvailable() >> true
+                    _ * getStatus() >> UpstreamAvailability.OK
                 }
         ]
         when:
@@ -259,12 +260,14 @@ class FilteredApisSpec extends Specification {
                 Mock([name: "fallback"], Upstream) {
                     _ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK
                     _ * isAvailable() >> true
+                    _ * getStatus() >> UpstreamAvailability.OK
                 }
         ]
         List<Upstream> secondary = [
                 Mock([name: "secondary"], Upstream) {
                     _ * getRole() >> UpstreamsConfig.UpstreamRole.SECONDARY
                     _ * isAvailable() >> true
+                    _ * getStatus() >> UpstreamAvailability.OK
                 }
         ]
         when:
@@ -287,4 +290,35 @@ class FilteredApisSpec extends Specification {
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
     }
+
+    def "Use LAGGING after OK"() {
+        setup:
+        List<Upstream> lagging = [
+                Mock([name: "lagging"], Upstream) {
+                    _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
+                    _ * isAvailable() >> true
+                    _ * getStatus() >> UpstreamAvailability.LAGGING
+                }
+        ]
+        List<Upstream> ok = [
+                Mock([name: "ok"], Upstream) {
+                    _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
+                    _ * isAvailable() >> true
+                    _ * getStatus() >> UpstreamAvailability.OK
+                }
+        ]
+        when:
+        def act = new FilteredApis(Chain.ETHEREUM,
+                [] + lagging + ok,
+                Selector.empty, 0, 2, 0)
+        act.request(4)
+        then:
+        StepVerifier.create(act)
+                .expectNext(ok[0]).as("Initial requests with ok")
+                .expectNext(lagging[0]).as("Initial requests with lagging")
+                .expectNext(ok[0]).as("retry requests with ok")
+                .expectNext(lagging[0]).as("retry requests with lagging")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
 }