solution: correctly account for upstream lag

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

@@ -15,6 +15,7 @@ class ChainUpstreams (
     private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
     private var seq = 0
     private var head: EthereumHead?
+    private var lagObserver: HeadLagObserver? = null
 
     init {
         head = updateHead()
@@ -28,10 +29,18 @@ class ChainUpstreams (
         if (current != null && Closeable::class.java.isAssignableFrom(current.javaClass)) {
             (current as Closeable).close()
         }
+        lagObserver?.close()
+        lagObserver = null
         return if (upstreams.size == 1) {
-            upstreams.first().getHead()
+            val upstream = upstreams.first()
+            upstream.setLag(0)
+            upstream.getHead()
         } else {
-            EthereumHeadMerge(upstreams.map { it.getHead() })
+            val newHead = EthereumHeadMerge(upstreams.map { it.getHead() })
+            val lagObserver = HeadLagObserver(newHead, upstreams)
+            lagObserver.start()
+            this.lagObserver = lagObserver
+            newHead
         }
     }
 
@@ -60,6 +69,13 @@ class ChainUpstreams (
         return head!!
     }
 
+    override fun setLag(lag: Long) {
+    }
+
+    override fun getLag(): Long {
+        return 0
+    }
+
     fun printStatus() {
         var height: Long? = null
         try {
@@ -73,8 +89,10 @@ class ChainUpstreams (
                 .groupBy { it }
                 .map { "${it.key.name}/${it.value.size}" }
                 .joinToString(",")
+        val lag = upstreams.map { it.getLag() }
+                .joinToString(", ")
 
-        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses")
+        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]")
     }
 
 

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

@@ -16,6 +16,7 @@ import reactor.core.publisher.toFlux
 import java.io.File
 import java.net.URI
 import java.util.*
+import java.util.concurrent.ConcurrentHashMap
 import javax.annotation.PostConstruct
 import kotlin.collections.HashMap
 
@@ -27,7 +28,7 @@ open class ConfiguredUpstreams(
 ) : Upstreams {
 
     private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
-    private val chainMapping = HashMap<Chain, ChainUpstreams>()
+    private val chainMapping = ConcurrentHashMap<Chain, ChainUpstreams>()
 
     private val chainNames = mapOf(
             "ethereum" to Chain.ETHEREUM,

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

@@ -0,0 +1,56 @@
+package io.emeraldpay.dshackle.upstream
+
+import reactor.core.publisher.Flux
+import reactor.core.publisher.TopicProcessor
+import java.util.concurrent.atomic.AtomicReference
+
+abstract class DefaultUpstream(
+        lag: Long,
+        avail: UpstreamAvailability
+) : Upstream {
+
+    constructor() : this(Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE)
+
+    private val status = AtomicReference(Status(lag, avail, statusByLag(lag, avail)))
+    private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
+
+    override fun getStatus(): UpstreamAvailability {
+        return status.get().status
+    }
+
+    fun setStatus(avail: UpstreamAvailability) {
+        status.updateAndGet { curr ->
+            Status(curr.lag, avail, statusByLag(curr.lag, avail))
+        }
+    }
+
+    fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
+        return if (proposed == UpstreamAvailability.OK) {
+            when {
+                lag > 6 -> UpstreamAvailability.SYNCING
+                lag > 1 -> UpstreamAvailability.LAGGING
+                else -> proposed
+            }
+        } else proposed
+    }
+
+    override fun observeStatus(): Flux<UpstreamAvailability> {
+        return Flux.from(statusStream)
+    }
+
+    override fun setLag(lag: Long) {
+        if (lag < 0) {
+            setLag(0)
+        } else {
+            status.updateAndGet { curr ->
+                Status(lag, curr.avail, statusByLag(lag, curr.avail))
+            }
+        }
+    }
+
+    override fun getLag(): Long {
+        return this.status.get().lag
+    }
+
+    class Status(val lag: Long, val avail: UpstreamAvailability, val status: UpstreamAvailability)
+}

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

@@ -7,6 +7,7 @@ import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.TopicProcessor
+import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicReference
 
 open class EthereumUpstream(
@@ -16,7 +17,7 @@ open class EthereumUpstream(
         private val options: UpstreamsConfig.Options,
         val node: NodeDetailsList.NodeDetails,
         private val targets: EthereumTargets
-): Upstream {
+): DefaultUpstream() {
 
     override fun getSupportedTargets(): Set<String> {
         return targets.getSupportedMethods()
@@ -33,34 +34,17 @@ open class EthereumUpstream(
     }
 
     private val validator = UpstreamValidator(this, options)
-    private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE)
-    private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
 
     init {
         log.info("Configured for ${chain.chainName}")
         api.upstream = this
 
         validator.start()
-                .subscribe {
+                .subscribe(this::setStatus)
-                    status.set(it)
-                    statusStream.onNext(it)
-                }
     }
 
     override fun isAvailable(matcher: Selector.Matcher): Boolean {
-        return status.get() == UpstreamAvailability.OK && matcher.matches(node.labels)
+        return getStatus() == UpstreamAvailability.OK && matcher.matches(node.labels)
-    }
-
-    override fun getStatus(): UpstreamAvailability {
-        return status.get()
-    }
-
-    fun setStatus(avail: UpstreamAvailability) {
-        status.set(avail)
-    }
-
-    override fun observeStatus(): Flux<UpstreamAvailability> {
-        return Flux.from(statusStream)
     }
 
     override fun getHead(): EthereumHead {
@@ -78,4 +62,5 @@ open class EthereumUpstream(
     override fun getOptions(): UpstreamsConfig.Options {
         return options
     }
+
 }

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

@@ -30,7 +30,7 @@ open class GrpcUpstream(
         private val objectMapper: ObjectMapper,
         private val options: UpstreamsConfig.Options,
         private val targets: EthereumTargets
-): Upstream {
+): DefaultUpstream() {
 
     constructor(chain: Chain, client: ReactorBlockchainGrpc.ReactorBlockchainStub, objectMapper: ObjectMapper, targets: EthereumTargets)
             : this(chain, client, objectMapper, UpstreamsConfig.Options.getDefaults(), targets)
@@ -123,11 +123,6 @@ open class GrpcUpstream(
         )
     }
 
-    private fun setStatus(value: UpstreamAvailability) {
-        status.set(value)
-        statusStream.onNext(value)
-    }
-
     fun getNodes(): NodeDetailsList {
         return nodes.get()
     }
@@ -144,14 +139,6 @@ open class GrpcUpstream(
         }
     }
 
-    override fun getStatus(): UpstreamAvailability {
-        return status.get()
-    }
-
-    override fun observeStatus(): Flux<UpstreamAvailability> {
-        return Flux.from(statusStream)
-    }
-
     override fun getHead(): EthereumHead {
         return head
     }

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

@@ -0,0 +1,77 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import org.slf4j.LoggerFactory
+import reactor.core.Disposable
+import reactor.core.publisher.Flux
+import reactor.core.publisher.toFlux
+import reactor.util.function.Tuple2
+import reactor.util.function.Tuples
+import java.io.Closeable
+import java.time.Duration
+
+class HeadLagObserver (
+        private val master: EthereumHead,
+        private val followers: Collection<Upstream>
+): Closeable {
+
+    private val log = LoggerFactory.getLogger(HeadLagObserver::class.java)
+
+    private var current: Disposable? = null
+
+    fun start() {
+        current = subscription().subscribe { }
+    }
+
+    private fun subscription(): Flux<Unit> {
+        return master.getFlux()
+                .flatMap(this::probeFollowers)
+                .map { item ->
+                    item.t2.setLag(item.t1)
+                }
+    }
+
+    fun probeFollowers(top: BlockJson<TransactionId>): Flux<Tuple2<Long, Upstream>> {
+        return followers.toFlux()
+                .parallel(followers.size)
+                .flatMap { mapLagging(top, it, getCurrentBlocks(it)) }
+                .sequential()
+                .onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) }
+    }
+
+    fun getCurrentBlocks(up: Upstream): Flux<BlockJson<TransactionId>> {
+        val head = up.getHead()
+        return Flux.concat(head.getHead(), head.getFlux())
+                .take(Duration.ofSeconds(1))
+    }
+
+    fun mapLagging(top: BlockJson<TransactionId>, up: Upstream, blocks: Flux<BlockJson<TransactionId>>): Flux<Tuple2<Long, Upstream>> {
+        return blocks
+                .map { extractDistance(top, it) }
+                .takeUntil{ lag -> lag <= 0L }
+                .map { Tuples.of(it, up) }
+                .doOnError { t ->
+                    log.warn("Failed to find distance for $up", t)
+                }
+    }
+
+    fun extractDistance(top: BlockJson<TransactionId>, curr: BlockJson<TransactionId>): Long {
+        return  when {
+            curr.number  > top.number -> if (curr.totalDifficulty >= top.totalDifficulty) 0 else forkDistance(top, curr)
+            curr.number == top.number -> if (curr.totalDifficulty == top.totalDifficulty) 0 else forkDistance(top, curr)
+            else -> top.number - curr.number
+        }
+    }
+
+    fun forkDistance(top: BlockJson<TransactionId>, curr: BlockJson<TransactionId>): Long {
+        //TODO look for common ancestor? though it may be a corruption
+        return 6
+    }
+
+    override fun close() {
+        current?.dispose()
+        current = null
+    }
+
+}

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

@@ -2,44 +2,32 @@ package io.emeraldpay.dshackle.upstream
 
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
+import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.util.concurrent.atomic.AtomicReference
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
 
 class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
 
-    private var head: Flux<BlockJson<TransactionId>> = Flux.empty<BlockJson<TransactionId>>()
+    private val result: AtomicReference<ByteArray> = AtomicReference()
-    private val lock = ReentrantLock()
-    private var resolved = false
-    private var result: ByteArray? = null
 
     override fun init(head: Head<BlockJson<TransactionId>>) {
-        this.head = head.getFlux()
     }
 
     override fun isResolved(): Boolean {
-        return resolved && result != null
+        return result.get() != null
     }
 
     override fun record(response: ByteArray, upstream: Upstream) {
-        Mono.from(head)
+        val lagging = upstream.getLag() > maxLag
-                .zipWith(upstream.getHead().getHead())
+        if (!lagging) {
-                .map {
+            result.set(response)
-                    val top = it.t1
+        }
-                    val current = it.t2
-                    return@map (top.number - current.number) < maxLag
-                }.subscribe { fresh ->
-                    if (fresh) {
-                        lock.withLock {
-                            result = response
-                            resolved = true
-                        }
-                    }
-                }
     }
 
     override fun getResult(): ByteArray {
-        return result!!
+        return result.get()
     }
 }

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

@@ -11,4 +11,6 @@ interface Upstream {
     fun getApi(matcher: Selector.Matcher): EthereumApi
     fun getOptions(): UpstreamsConfig.Options
     fun getSupportedTargets(): Set<String>
+    fun setLag(lag: Long)
+    fun getLag(): Long
 }

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

@@ -32,6 +32,7 @@ class FilteringApiIteratorSpec extends Specification {
         }
         def matcher = new Selector.LabelMatcher("test", ["foo"])
         upstreams.forEach {
+            it.setLag(0)
             it.setStatus(UpstreamAvailability.OK)
         }
         when:

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

@@ -56,6 +56,7 @@ class GrpcUpstreamSpec extends Specification {
             }
         })
         def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets)
+        upstream.setLag(0)
         when:
         upstream.connect()
         def h = upstream.head.head.block(Duration.ofSeconds(1))
@@ -111,6 +112,7 @@ class GrpcUpstreamSpec extends Specification {
             }
         })
         def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets)
+        upstream.setLag(0)
         when:
         upstream.connect()
         finished.get()
@@ -167,6 +169,7 @@ class GrpcUpstreamSpec extends Specification {
             }
         })
         def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets)
+        upstream.setLag(0)
         when:
         upstream.connect()
         finished.get()

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

@@ -0,0 +1,116 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.infinitape.etherjar.rpc.json.BlockJson
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.TopicProcessor
+import reactor.test.StepVerifier
+import reactor.util.function.Tuples
+import spock.lang.Specification
+
+import java.time.Duration
+
+class HeadLagObserverSpec extends Specification {
+
+    def "Updates lag distance"() {
+        setup:
+        EthereumHead master = Mock()
+
+        EthereumHead head1 = Mock()
+        EthereumHead head2 = Mock()
+
+        Upstream up1 = Mock {
+            _ * getHead() >> head1
+        }
+        Upstream up2 = Mock {
+            _ * getHead() >> head2
+        }
+
+        def blocks = [100, 101, 102].collect { i ->
+            return new BlockJson().with {
+                it.number = i
+                it.totalDifficulty = 2000 + i
+                return it
+            }
+        }
+
+        def masterBus = TopicProcessor.create()
+
+        1 * master.getFlux() >> Flux.from(masterBus)
+        1 * head1.getHead() >> Mono.just(blocks[1])
+        1 * head1.getFlux() >> Flux.just(blocks[2])
+                .delaySubscription(Duration.ofSeconds(1))
+        1 * head2.getHead() >> Mono.just(blocks[0])
+        1 * head2.getFlux() >> Flux.just(blocks[1])
+                .delaySubscription(Duration.ofMillis(100))
+        1 * up1.setLag(0)
+        1 * up2.setLag(1)
+        1 * up2.setLag(0)
+
+        HeadLagObserver observer = new HeadLagObserver(master, [up1, up2])
+        when:
+        def act = observer.subscription().take(Duration.ofMillis(1200))
+
+        then:
+        StepVerifier.create(act)
+            .then { masterBus.onNext(blocks[1]) }
+            .expectNextCount(3)
+            .verifyComplete()
+    }
+
+    def "Probes until there is no difference"() {
+        setup:
+        EthereumHead master = Mock()
+        HeadLagObserver observer = new HeadLagObserver(master, [])
+        Upstream up = Mock()
+
+        def blocks = [100, 101, 102].collect { i ->
+            return new BlockJson().with {
+                it.number = i
+                it.totalDifficulty = 2000 + i
+                return it
+            }
+        }
+
+        def upblocks = Flux.fromIterable(blocks)
+        when:
+        def act = observer.mapLagging(blocks[2], up, upblocks)
+        then:
+        StepVerifier.create(act)
+                .expectNext(Tuples.of(2L, up))
+                .expectNext(Tuples.of(1L, up))
+                .expectNext(Tuples.of(0L, up))
+                .verifyComplete()
+    }
+
+    def "Correct distance"() {
+        setup:
+        EthereumHead master = Mock()
+        HeadLagObserver observer = new HeadLagObserver(master, [])
+        expect:
+        def top = new BlockJson().with {
+            it.number = topHeight
+            it.totalDifficulty = topDiff
+            return it
+        }
+        def curr = new BlockJson().with {
+            it.number = currHeight
+            it.totalDifficulty = currDiff
+            return it
+        }
+        delta as Long == observer.extractDistance(top, curr)
+        where:
+        topHeight | topDiff | currHeight | currDiff | delta
+        100       | 1000    | 100        | 1000     | 0
+        101       | 1010    | 100        | 1000     | 1
+        102       | 1020    | 100        | 1000     | 2
+        103       | 1030    | 100        | 1000     | 3
+        150       | 1500    | 100        | 1000     | 50
+
+        100       | 1000    | 101        | 1010     | 0
+        100       | 1000    | 102        | 1020     | 0
+        100       | 1000    | 100        | 1010     | 6
+        100       | 1100    | 100        | 1000     | 6
+
+    }
+}

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

@@ -0,0 +1,47 @@
+package io.emeraldpay.dshackle.upstream
+
+import spock.lang.Specification
+
+class NotLaggingQuorumSpec extends Specification {
+
+    def "Resolves if no lag"() {
+        setup:
+        def up = Mock(Upstream)
+        def value = "foo".getBytes()
+        def quorum = new NotLaggingQuorum(1)
+
+        when:
+        quorum.record(value, up)
+        then:
+        1 * up.getLag() >> 0
+        quorum.isResolved()
+        quorum.result == value
+    }
+
+    def "Resolves if ok lag"() {
+        setup:
+        def up = Mock(Upstream)
+        def value = "foo".getBytes()
+        def quorum = new NotLaggingQuorum(1)
+
+        when:
+        quorum.record(value, up)
+        then:
+        1 * up.getLag() >> 1
+        quorum.isResolved()
+        quorum.result == value
+    }
+
+    def "Ignores if lags"() {
+        setup:
+        def up = Mock(Upstream)
+        def value = "foo".getBytes()
+        def quorum = new NotLaggingQuorum(1)
+
+        when:
+        quorum.record(value, up)
+        then:
+        1 * up.getLag() >> 2
+        !quorum.isResolved()
+    }
+}