StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

We should always send only best status form Multistream.observeStatus

Browse Source
This commit is contained in:
a10zn8
2023-06-02 14:20:32 +04:00
parent f626cf61cb
commit 64705f612d
2 changed files with 40 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -40,10 +40,8 @@ import reactor.core.publisher.Mono
import reactor.core.publisher.Sinks import reactor.core.publisher.Sinks
import reactor.util.function.Tuples import reactor.util.function.Tuples
import java.time.Duration import java.time.Duration
import java.time.Instant import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.locks.ReentrantLock
import java.util.function.Predicate
import kotlin.concurrent.withLock import kotlin.concurrent.withLock
/** /**
@@ -232,8 +230,8 @@ abstract class Multistream(
).map { UpstreamStatus(up, it) } ).map { UpstreamStatus(up, it) }
} }
return Flux.merge(upstreamsFluxes) return Flux.merge(upstreamsFluxes)
.filter(FilterBestAvailability()) .map(FilterBestAvailability())
.map { it.status } .distinct()
} }
override fun isAvailable(): Boolean { override fun isAvailable(): Boolean {
@@ -452,24 +450,13 @@ abstract class Multistream(
// -------------------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------------------
class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability)
class FilterBestAvailability : Predicate<UpstreamStatus> { class FilterBestAvailability : java.util.function.Function<UpstreamStatus, UpstreamAvailability> {
private val lastRef = AtomicReference<UpstreamStatus>() val map = ConcurrentHashMap<String, UpstreamAvailability>()
override fun apply(t: UpstreamStatus): UpstreamAvailability {
override fun test(t: UpstreamStatus): Boolean { map[t.upstream.getId()] = t.status
val curr = lastRef.updateAndGet { last -> return map.values.min()
val changed = last == null ||
t.status < last.status ||
(last.upstream == t.upstream && t.status != last.status) ||
last.ts.isBefore(t.ts - Duration.ofSeconds(60))
if (changed) {
t
} else {
last
}
}
return curr == t
} }
} }

67
src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
View File

@@ -70,87 +70,83 @@ class MultistreamSpec extends Specification {
def up = TestingCommons.upstream() def up = TestingCommons.upstream()
def filter = new Multistream.FilterBestAvailability() def filter = new Multistream.FilterBestAvailability()
def update1 = new Multistream.UpstreamStatus( def update1 = new Multistream.UpstreamStatus(
up, UpstreamAvailability.LAGGING, Instant.now() up, UpstreamAvailability.LAGGING
) )
when: when:
def act = filter.test(update1) def status = filter.apply(update1)
then: then:
act status == UpstreamAvailability.LAGGING
} }
def "Filter Best Status accepts better input"() { def "Filter Best Status accepts better input"() {
setup: setup:
def up1 = TestingCommons.upstream("test-1") def up1 = TestingCommons.upstream("test-1")
def up2 = TestingCommons.upstream("test-2") def up2 = TestingCommons.upstream("test-2")
def time0 = Instant.now() - Duration.ofSeconds(60)
def filter = new Multistream.FilterBestAvailability() def filter = new Multistream.FilterBestAvailability()
def update0 = new Multistream.UpstreamStatus( def update0 = new Multistream.UpstreamStatus(
up1, UpstreamAvailability.LAGGING, time0 up1, UpstreamAvailability.LAGGING
) )
def update1 = new Multistream.UpstreamStatus( def update1 = new Multistream.UpstreamStatus(
up2, UpstreamAvailability.OK, time0 up2, UpstreamAvailability.OK
) )
when: when:
filter.test(update0) filter.apply(update0)
def act = filter.test(update1) def status = filter.apply(update1)
then: then:
act status == UpstreamAvailability.OK
} }
def "Filter Best Status declines worse input"() { def "Filter Best Status declines worse input"() {
setup: setup:
def up1 = TestingCommons.upstream("test-1") def up1 = TestingCommons.upstream("test-1")
def up2 = TestingCommons.upstream("test-2") def up2 = TestingCommons.upstream("test-2")
def time0 = Instant.now() - Duration.ofSeconds(60)
def filter = new Multistream.FilterBestAvailability() def filter = new Multistream.FilterBestAvailability()
def update0 = new Multistream.UpstreamStatus( def update0 = new Multistream.UpstreamStatus(
up1, UpstreamAvailability.LAGGING, time0 up1, UpstreamAvailability.LAGGING
) )
def update1 = new Multistream.UpstreamStatus( def update1 = new Multistream.UpstreamStatus(
up2, UpstreamAvailability.IMMATURE, time0 up2, UpstreamAvailability.IMMATURE
) )
when: when:
filter.test(update0) filter.apply(update0)
def act = filter.test(update1) def status = filter.apply(update1)
then: then:
!act status == UpstreamAvailability.LAGGING
} }
def "Filter Best Status accepts worse input from same upstream"() { def "Filter Best Status accepts worse input from same upstream"() {
setup: setup:
def up = TestingCommons.upstream("test-1") def up = TestingCommons.upstream("test-1")
def time0 = Instant.now() - Duration.ofSeconds(60)
def filter = new Multistream.FilterBestAvailability() def filter = new Multistream.FilterBestAvailability()
def update0 = new Multistream.UpstreamStatus( def update0 = new Multistream.UpstreamStatus(
up, UpstreamAvailability.LAGGING, time0 up, UpstreamAvailability.LAGGING
) )
def update1 = new Multistream.UpstreamStatus( def update1 = new Multistream.UpstreamStatus(
up, UpstreamAvailability.IMMATURE, time0 up, UpstreamAvailability.IMMATURE
) )
when: when:
filter.test(update0) filter.apply(update0)
def act = filter.test(update1) def status = filter.apply(update1)
then: then:
act status == UpstreamAvailability.IMMATURE
} }
def "Filter Best Status accepts any input if existing is outdated"() { def "Filter Best Status accepts any input if existing is outdated"() {
setup: setup:
def up1 = TestingCommons.upstream("test-1") def up1 = TestingCommons.upstream("test-1")
def up2 = TestingCommons.upstream("test-2") def up2 = TestingCommons.upstream("test-2")
def time0 = Instant.now() - Duration.ofSeconds(90)
def filter = new Multistream.FilterBestAvailability() def filter = new Multistream.FilterBestAvailability()
def update0 = new Multistream.UpstreamStatus( def update0 = new Multistream.UpstreamStatus(
up1, UpstreamAvailability.OK, time0 up1, UpstreamAvailability.OK
) )
def update1 = new Multistream.UpstreamStatus( def update1 = new Multistream.UpstreamStatus(
up2, UpstreamAvailability.IMMATURE, time0 + Duration.ofSeconds(65) up2, UpstreamAvailability.IMMATURE
) )
when: when:
filter.test(update0) filter.apply(update0)
def act = filter.test(update1) def status = filter.apply(update1)
then: then:
act status == UpstreamAvailability.OK
} }
def "Filter Best Status declines same status"() { def "Filter Best Status declines same status"() {
@@ -158,28 +154,27 @@ class MultistreamSpec extends Specification {
def up1 = TestingCommons.upstream("test-1") def up1 = TestingCommons.upstream("test-1")
def up2 = TestingCommons.upstream("test-2") def up2 = TestingCommons.upstream("test-2")
def up3 = TestingCommons.upstream("test-3") def up3 = TestingCommons.upstream("test-3")
def time0 = Instant.now() - Duration.ofSeconds(60)
def filter = new Multistream.FilterBestAvailability() def filter = new Multistream.FilterBestAvailability()
def update0 = new Multistream.UpstreamStatus( def update0 = new Multistream.UpstreamStatus(
up1, UpstreamAvailability.OK, time0 up1, UpstreamAvailability.OK
) )
def update1 = new Multistream.UpstreamStatus( def update1 = new Multistream.UpstreamStatus(
up2, UpstreamAvailability.OK, time0 up2, UpstreamAvailability.OK
) )
def update2 = new Multistream.UpstreamStatus( def update2 = new Multistream.UpstreamStatus(
up3, UpstreamAvailability.OK, time0 + Duration.ofSeconds(10) up3, UpstreamAvailability.OK
) )
when: when:
filter.test(update0) filter.apply(update0)
def act = filter.test(update1) def status = filter.apply(update1)
then: then:
!act status == UpstreamAvailability.OK
when: when:
act = filter.test(update2) status = filter.apply(update2)
then: then:
!act status == UpstreamAvailability.OK
} }