problem: incorrect status updates

emeraldpay · Sep 6, 2021 · f6134ed · f6134ed
1 parent 8c7bb01
commit f6134ed
Show file tree

Hide file tree

Showing 3 changed files with 126 additions and 2 deletions.
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
@@ -296,9 +296,9 @@ abstract class Multistream(
         override fun test(t: UpstreamStatus): Boolean {
             val curr = lastRef.updateAndGet { last ->
                 val changed = last == null
-                        || t.status > last.status
+                        || t.status < last.status
                         || (last.upstream == t.upstream && t.status != last.status)
-                        || last.ts.isBefore(Instant.now() - Duration.ofSeconds(60))
+                        || last.ts.isBefore(t.ts - Duration.ofSeconds(60))
                 if (changed) {
                     t
                 } else {

diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
@@ -54,6 +54,10 @@ class TestingCommons {
         return new EthereumUpstreamMock(Chain.ETHEREUM, api())
     }
 
+    static EthereumUpstreamMock upstream(String id) {
+        return new EthereumUpstreamMock(id, Chain.ETHEREUM, api())
+    }
+
     static EthereumUpstreamMock upstream(String id, Reader<JsonRpcRequest, JsonRpcResponse> api) {
         return new EthereumUpstreamMock(id, Chain.ETHEREUM, api)
     }

diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
@@ -25,6 +25,9 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.grpc.Chain
 import spock.lang.Specification
 
+import java.time.Duration
+import java.time.Instant
+
 class MultistreamSpec extends Specification {
 
     def "Aggregates methods"() {
@@ -43,4 +46,121 @@ class MultistreamSpec extends Specification {
         act.getQuorumFor("eth_test2") instanceof AlwaysQuorum
         act.getQuorumFor("eth_test3") instanceof AlwaysQuorum
     }
+
+    def "Filter Best Status accepts any input when none available "() {
+        setup:
+        def up = TestingCommons.upstream()
+        def filter = new Multistream.FilterBestAvailability()
+        def update1 = new Multistream.UpstreamStatus(
+                up, UpstreamAvailability.LAGGING, Instant.now()
+        )
+        when:
+        def act = filter.test(update1)
+        then:
+        act
+    }
+
+    def "Filter Best Status accepts better input"() {
+        setup:
+        def up1 = TestingCommons.upstream("1")
+        def up2 = TestingCommons.upstream("2")
+        def time0 = Instant.now() - Duration.ofSeconds(60)
+        def filter = new Multistream.FilterBestAvailability()
+        def update0 = new Multistream.UpstreamStatus(
+                up1, UpstreamAvailability.LAGGING, time0
+        )
+        def update1 = new Multistream.UpstreamStatus(
+                up2, UpstreamAvailability.OK, time0
+        )
+        when:
+        filter.test(update0)
+        def act = filter.test(update1)
+        then:
+        act
+    }
+
+    def "Filter Best Status declines worse input"() {
+        setup:
+        def up1 = TestingCommons.upstream("1")
+        def up2 = TestingCommons.upstream("2")
+        def time0 = Instant.now() - Duration.ofSeconds(60)
+        def filter = new Multistream.FilterBestAvailability()
+        def update0 = new Multistream.UpstreamStatus(
+                up1, UpstreamAvailability.LAGGING, time0
+        )
+        def update1 = new Multistream.UpstreamStatus(
+                up2, UpstreamAvailability.IMMATURE, time0
+        )
+        when:
+        filter.test(update0)
+        def act = filter.test(update1)
+        then:
+        !act
+    }
+
+    def "Filter Best Status accepts worse input from same upstream"() {
+        setup:
+        def up = TestingCommons.upstream("1")
+        def time0 = Instant.now() - Duration.ofSeconds(60)
+        def filter = new Multistream.FilterBestAvailability()
+        def update0 = new Multistream.UpstreamStatus(
+                up, UpstreamAvailability.LAGGING, time0
+        )
+        def update1 = new Multistream.UpstreamStatus(
+                up, UpstreamAvailability.IMMATURE, time0
+        )
+        when:
+        filter.test(update0)
+        def act = filter.test(update1)
+        then:
+        act
+    }
+
+    def "Filter Best Status accepts any input if existing is outdated"() {
+        setup:
+        def up1 = TestingCommons.upstream("1")
+        def up2 = TestingCommons.upstream("2")
+        def time0 = Instant.now() - Duration.ofSeconds(90)
+        def filter = new Multistream.FilterBestAvailability()
+        def update0 = new Multistream.UpstreamStatus(
+                up1, UpstreamAvailability.OK, time0
+        )
+        def update1 = new Multistream.UpstreamStatus(
+                up2, UpstreamAvailability.IMMATURE, time0 + Duration.ofSeconds(65)
+        )
+        when:
+        filter.test(update0)
+        def act = filter.test(update1)
+        then:
+        act
+    }
+
+    def "Filter Best Status declines same status"() {
+        setup:
+        def up1 = TestingCommons.upstream("1")
+        def up2 = TestingCommons.upstream("2")
+        def up3 = TestingCommons.upstream("3")
+        def time0 = Instant.now() - Duration.ofSeconds(60)
+        def filter = new Multistream.FilterBestAvailability()
+        def update0 = new Multistream.UpstreamStatus(
+                up1, UpstreamAvailability.OK, time0
+        )
+        def update1 = new Multistream.UpstreamStatus(
+                up2, UpstreamAvailability.OK, time0
+        )
+        def update2 = new Multistream.UpstreamStatus(
+                up3, UpstreamAvailability.OK, time0 + Duration.ofSeconds(10)
+        )
+
+        when:
+        filter.test(update0)
+        def act = filter.test(update1)
+        then:
+        !act
+
+        when:
+        act = filter.test(update2)
+        then:
+        !act
+    }
 }