solution: stream updates to blockchain status

src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt

@@ -1,10 +1,14 @@
 package io.emeraldpay.dshackle.rpc
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.Upstreams
+import io.grpc.StatusRuntimeException
 import io.grpc.stub.StreamObserver
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
+import reactor.core.Disposable
 
 @Service
 class SubscribeStatus(
@@ -13,6 +17,24 @@ class SubscribeStatus(
 
     fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver<BlockchainOuterClass.ChainStatus>) {
         upstreams.getAvailable().forEach { chain ->
+            var d: Disposable? = null
+            d = upstreams.getUpstream(chain)?.observeStatus()?.subscribe { availability ->
+                val chainStatus = BlockchainOuterClass.ChainStatus.newBuilder()
+                        .setChain(Common.ChainRef.forNumber(chain.id))
+                        .setAvailable(availability == UpstreamAvailability.OK)
+                        .setQuorum(0)
+                if (availability == UpstreamAvailability.OK) {
+                    upstreams.getUpstream(chain)?.getOptions()?.let { opts ->
+                        chainStatus.setQuorum(opts.quorum)
+                    }
+                }
+                try {
+                    responseObserver.onNext(chainStatus.build())
+                } catch (e: StatusRuntimeException) {
+                    // gRPC channel was closed
+                    d?.dispose()
+                }
+            }
         }
     }
 

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

@@ -1,12 +1,44 @@
 package io.emeraldpay.dshackle.upstream
 
-abstract class AggregatedUpstreams {
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import reactor.core.publisher.Flux
+import java.time.Duration
+import java.time.Instant
+import java.util.concurrent.atomic.AtomicReference
+import java.util.function.Predicate
+
+abstract class AggregatedUpstreams: Upstream {
 
     abstract fun getAll(): List<Upstream>
     abstract fun addUpstream(upstream: Upstream)
     abstract fun getApis(quorum: Int): Iterator<EthereumApi>
-    abstract fun getApi(): EthereumApi
+
-    abstract fun getHead(): EthereumHead
+    override fun observeStatus(): Flux<UpstreamAvailability> {
+        val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } }
+        return Flux.merge(upstreamsFluxes)
+                .filter(FilterBestAvailability())
+                .map { it.status }
+    }
+
+    override fun isAvailable(): Boolean {
+        return getAll().any { it.isAvailable() }
+    }
+
+    override fun getStatus(): UpstreamAvailability {
+        val upstreams = getAll()
+        return if (upstreams.isEmpty()) UpstreamAvailability.UNAVAILABLE
+        else upstreams.map { it.getStatus() }.min()!!
+    }
+
+    override fun getOptions(): UpstreamsConfig.Options {
+        val options = UpstreamsConfig.Options()
+        options.quorum = getAll().filter {
+            it.getStatus() == UpstreamAvailability.OK
+        }.sumBy {
+            it.getOptions().quorum
+        }
+        return options
+    }
 
     class SingleApi(
             private val quorumApi: QuorumApi
@@ -47,6 +79,23 @@ abstract class AggregatedUpstreams {
             }
             throw IllegalStateException("No upstream API available")
         }
+    }
 
+    class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())
+
+    class FilterBestAvailability(): Predicate<UpstreamStatus> {
+        private val lastRef = AtomicReference<UpstreamStatus>()
+
+        override fun test(t: UpstreamStatus): Boolean {
+            val last = lastRef.get()
+            val changed = last == null
+                    || t.status > last.status
+                    || (last.upstream == t.upstream && t.status != last.status)
+                    || last.ts.isBefore(Instant.now() - Duration.ofSeconds(60))
+            if (changed) {
+                lastRef.set(t)
+            }
+            return changed
+        }
     }
 }

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

@@ -1,5 +1,6 @@
 package io.emeraldpay.dshackle.upstream
 
+import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import java.io.Closeable
@@ -57,9 +58,9 @@ class ChainUpstreams (
     }
 
     fun printStatus() {
-        var height: Long = -1
+        var height: Long? = null
         try {
-            height = head!!.getHead().block(Duration.ofSeconds(1))?.number ?: -1
+            height = head!!.getHead().block(Duration.ofSeconds(1))?.number
         } catch (e: IllegalStateException) {
             //timout
         } catch (e: Exception) {
@@ -70,7 +71,7 @@ class ChainUpstreams (
                 .map { "${it.key.name}/${it.value.size}" }
                 .joinToString(",")
 
-        log.info("State of ${chain.chainCode}: height=$height, status=$statuses")
+        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses")
     }
 
 

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

@@ -16,6 +16,7 @@ import reactor.core.publisher.Mono
 import reactor.core.publisher.toFlux
 import reactor.util.function.Tuple3
 import reactor.util.function.Tuples
+import java.time.Duration
 import java.util.concurrent.CompletableFuture
 import java.util.function.Function
 
@@ -97,6 +98,7 @@ class EthereumGrpcTransport(
                         bi.onError(RpcException(-32603, "RPC response not received"))
                     }
                 }
+                .timeout(Duration.ofSeconds(15))
                 .toFuture()
     }
 }

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

@@ -2,7 +2,11 @@ package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.grpc.Chain
+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.TopicProcessor
 import java.util.concurrent.atomic.AtomicReference
 
 class EthereumUpstream(
@@ -24,6 +28,7 @@ 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}")
@@ -31,6 +36,7 @@ class EthereumUpstream(
         validator.start()
                 .subscribe {
                     status.set(it)
+                    statusStream.onNext(it)
                 }
     }
 
@@ -42,6 +48,10 @@ class EthereumUpstream(
         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/GrpcUpstream.kt

@@ -38,6 +38,7 @@ open class GrpcUpstream(
     private var status = AtomicReference<UpstreamAvailability>(UpstreamAvailability.UNAVAILABLE)
     private val head = Head(this)
     private val api: EthereumApi
+    private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
 
     init {
         val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper)
@@ -52,7 +53,7 @@ open class GrpcUpstream(
                 .toMono()
 
         val retry: Function<Flux<BlockchainOuterClass.ChainHead>, Flux<BlockchainOuterClass.ChainHead>> = Function {
-            status.set(UpstreamAvailability.UNAVAILABLE)
+            setStatus(UpstreamAvailability.UNAVAILABLE)
             client.subscribeHead(chainRef)
         }
 
@@ -80,14 +81,14 @@ open class GrpcUpstream(
                     log.debug("New block ${block.number} on ${chain}")
                     headBlock.set(block)
                     streamBlocks.onNext(block)
-                    status.set(UpstreamAvailability.OK)
+                    setStatus(UpstreamAvailability.OK)
                 }
     }
 
     fun init(conf: BlockchainOuterClass.DescribeChain) {
         val available = conf.available
         val quorum = conf.quorum
-        status.set(
+        setStatus(
                 if (available && quorum > 0) UpstreamAvailability.OK else UpstreamAvailability.UNAVAILABLE
         )
     }
@@ -95,11 +96,15 @@ open class GrpcUpstream(
     fun onStatus(value: BlockchainOuterClass.ChainStatus) {
         val available = value.available
         val quorum = value.quorum
-        status.set(
+        setStatus(
                 if (available && quorum > 0) UpstreamAvailability.OK else UpstreamAvailability.UNAVAILABLE
         )
     }
 
+    private fun setStatus(value: UpstreamAvailability) {
+        status.set(value)
+        statusStream.onNext(value)
+    }
     // ------------------------------------------------------------------------------------------
 
 
@@ -111,6 +116,10 @@ open class GrpcUpstream(
         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/Upstream.kt

@@ -1,10 +1,12 @@
 package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.config.UpstreamsConfig
+import reactor.core.publisher.Flux
 
 interface Upstream {
     fun isAvailable(): Boolean
     fun getStatus(): UpstreamAvailability
+    fun observeStatus(): Flux<UpstreamAvailability>
     fun getHead(): EthereumHead
     fun getApi(): EthereumApi
     fun getOptions(): UpstreamsConfig.Options

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

@@ -43,7 +43,7 @@ class EthereumGrpcTransportSpec extends Specification {
         def status = transport.execute(batch.items).get()
 
         then:
-        1 * otherSideUpstreams.ethereumUpstream(Chain.ETHEREUM) >> otherSideAggr
+        1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr
         1 * otherSideAggr.api >> otherSideApi
         status.failed == 0
         status.succeed == 1
@@ -87,7 +87,7 @@ class EthereumGrpcTransportSpec extends Specification {
         def status = transport.execute(batch.items).get()
 
         then:
-        1 * otherSideUpstreams.ethereumUpstream(Chain.ETHEREUM) >> otherSideAggr
+        1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr
         1 * otherSideAggr.api >> otherSideApi
         status.failed == 0
         status.succeed == 2

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

@@ -0,0 +1,36 @@
+package io.emeraldpay.dshackle.upstream
+
+import spock.lang.Specification
+
+class UpstreamAvailabilitySpec extends Specification {
+
+    def "Defined order"() {
+        expect:
+        UpstreamAvailability.OK.compareTo(UpstreamAvailability.IMMATURE) < 0
+        UpstreamAvailability.IMMATURE.compareTo(UpstreamAvailability.SYNCING) < 0
+        UpstreamAvailability.SYNCING.compareTo(UpstreamAvailability.LAGGING) < 0
+        UpstreamAvailability.LAGGING.compareTo(UpstreamAvailability.UNAVAILABLE) < 0
+
+        UpstreamAvailability.OK.compareTo(UpstreamAvailability.UNAVAILABLE) < 0
+        UpstreamAvailability.UNAVAILABLE.compareTo(UpstreamAvailability.OK) > 0
+    }
+
+    def "Sort array"() {
+        setup:
+        def items = [UpstreamAvailability.OK, UpstreamAvailability.SYNCING, UpstreamAvailability.IMMATURE, UpstreamAvailability.OK]
+        when:
+        Collections.sort(items)
+        then:
+        items == [UpstreamAvailability.OK, UpstreamAvailability.OK, UpstreamAvailability.IMMATURE, UpstreamAvailability.SYNCING]
+    }
+
+    def "First is most available"() {
+        expect:
+        [UpstreamAvailability.UNAVAILABLE, UpstreamAvailability.OK].toSorted().first() == UpstreamAvailability.OK
+        [UpstreamAvailability.OK, UpstreamAvailability.UNAVAILABLE].toSorted().first() == UpstreamAvailability.OK
+        [UpstreamAvailability.OK, UpstreamAvailability.LAGGING].toSorted().first() == UpstreamAvailability.OK
+        [UpstreamAvailability.LAGGING, UpstreamAvailability.UNAVAILABLE].toSorted().first() == UpstreamAvailability.LAGGING
+        [UpstreamAvailability.OK, UpstreamAvailability.SYNCING].toSorted().first() == UpstreamAvailability.OK
+        [UpstreamAvailability.SYNCING, UpstreamAvailability.UNAVAILABLE].toSorted().first() == UpstreamAvailability.SYNCING
+    }
+}