diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index a0ac5ef6..47960364 100644 --- 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 index 2266dfd6..ff8a480f 100644 --- 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 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 index 39298b0a..b2e8f08f 100644 --- 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 + } }