From 6d78b4927dca03c80ec88c80c4d30b68512269c5 Mon Sep 17 00:00:00 2001
From: a10zn8 <a1.cyb3r@gmail.com>
Date: Mon, 13 Feb 2023 18:48:01 +0400
Subject: [PATCH] Adjustments in availability determination: - now isAvailable
 return true in case upstreams are OK or LAGGING - All upstreams in Filtered
 API now sorted by status

---
 .../dshackle/upstream/DefaultUpstream.kt      |  2 +-
 .../dshackle/upstream/FilteredApis.kt         |  6 ++--
 .../dshackle/upstream/FilteredApisSpec.groovy | 34 +++++++++++++++++++
 3 files changed, 38 insertions(+), 4 deletions(-)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
index 46ed8e42..4b3d1624 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
+++ b/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) {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt
index c9173c78..0e370c70 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt
+++ b/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 second = Flux.fromIterable(secondaryUpstreams)
+        val first = Flux.fromIterable(primaryUpstreams.sortedBy { it.getStatus().grpcId })
+        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
diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
index c3c289be..100f72ae 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
+++ b/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))
+    }
 }