solution: verify upstream status

src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java

@@ -18,7 +18,7 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
-public class Upstreams {
+public class UpstreamsConfig {
 
     private String version;
     private List<DefaultOptions> defaultOptions;
@@ -49,9 +49,9 @@ public class Upstreams {
     }
 
     public static class Options {
-        private Boolean disableSyncing = true;
+        private Boolean disableSyncing;
-        private Integer minPeers = 1;
+        private Integer minPeers;
-        private Integer quorum = 1;
+        private Integer quorum;
 
         public Boolean getDisableSyncing() {
             return disableSyncing;
@@ -66,6 +66,9 @@ public class Upstreams {
         }
 
         public void setMinPeers(Integer minPeers) {
+            if (minPeers < 0) {
+                throw new IllegalArgumentException("minPeers must be positive number");
+            }
             this.minPeers = minPeers;
         }
 
@@ -74,8 +77,31 @@ public class Upstreams {
         }
 
         public void setQuorum(Integer quorum) {
+            if (quorum < 0) {
+                throw new IllegalArgumentException("quorum must be positive number");
+            }
             this.quorum = quorum;
         }
+
+        public Options merge(Options additional) {
+            if (additional == null) {
+                return this;
+            }
+            Options copy = new Options();
+            copy.setDisableSyncing(this.disableSyncing != null ? this.disableSyncing : additional.disableSyncing);
+            copy.setMinPeers(this.minPeers != null ? this.minPeers : additional.minPeers);
+            copy.setQuorum(this.quorum != null ? this.quorum : additional.quorum);
+            return copy;
+        }
+
+        public static Options getDefaults() {
+            Options options = new Options();
+            options.setDisableSyncing(true);
+            options.setMinPeers(1);
+            options.setQuorum(1);
+            return options;
+        }
+
     }
 
     public static class OptionsYaml extends TypeDescription {
@@ -154,6 +180,7 @@ public class Upstreams {
             this.endpoints = endpoints;
         }
 
+        @Nullable
         public Options getOptions() {
             return options;
         }

src/main/kotlin/io/emeraldpay/dshackle/config/Configuration.kt

@@ -1,4 +0,0 @@
-package io.emeraldpay.dshackle.config
-
-class Configuration {
-}

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -0,0 +1,17 @@
+package io.emeraldpay.dshackle.config
+
+import org.yaml.snakeyaml.Yaml
+import java.io.InputStream
+
+class UpstreamsConfigReader {
+
+    fun read(input: InputStream): UpstreamsConfig {
+        val yaml = Yaml()
+        yaml.addTypeDescription(UpstreamsConfig.EndpointTypeYaml())
+        yaml.addTypeDescription(UpstreamsConfig.OptionsYaml())
+        yaml.addTypeDescription(UpstreamsConfig.AuthYaml())
+        return yaml.loadAs(input, UpstreamsConfig::class.java)
+    }
+
+
+}

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsReader.kt

@@ -1,17 +0,0 @@
-package io.emeraldpay.dshackle.config
-
-import org.yaml.snakeyaml.Yaml
-import java.io.InputStream
-
-class UpstreamsReader {
-
-    fun read(input: InputStream): Upstreams {
-        val yaml = Yaml()
-        yaml.addTypeDescription(Upstreams.EndpointTypeYaml())
-        yaml.addTypeDescription(Upstreams.OptionsYaml())
-        yaml.addTypeDescription(Upstreams.AuthYaml())
-        return yaml.loadAs(input, Upstreams::class.java)
-    }
-
-
-}

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

@@ -1,13 +1,18 @@
 package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import org.slf4j.LoggerFactory
 import java.lang.IllegalStateException
+import java.time.Duration
 
 class ChainConnect(
         val chain: Chain,
         val upstreams: List<Upstream>
         ) {
 
+    private val log = LoggerFactory.getLogger(ChainConnect::class.java)
     private var seq = 0
 
     val head: EthereumHead = if (upstreams.size == 1) {
@@ -29,6 +34,19 @@ class ChainConnect(
         return QuorumApi(upstreams, 1, seq)
     }
 
+    fun printStatus() {
+        var height: Long = -1
+        try {
+            height = head.getHead().block(Duration.ofSeconds(1))?.number ?: -1
+        } catch (e: Exception) { }
+        val statuses = upstreams.map { it.getStatus() }
+                .groupBy { it }
+                .map { "${it.key.name}/${it.value.size}" }
+                .joinToString(",")
+
+        log.info("State of ${chain.chainCode}: height=$height, status=$statuses")
+    }
+
     class SingleApi(
             val quorumApi: QuorumApi
     ): Iterator<EthereumApi> {

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

@@ -20,12 +20,17 @@ class EthereumWs(
             .builder<BlockJson<TransactionId>>()
             .name("new-blocks")
             .build()
-    private val head = AtomicReference<BlockJson<TransactionId>>(null);
+    private val head = AtomicReference<BlockJson<TransactionId>>(null)
 
     fun connect() {
         log.info("Connecting to WebSocket: $uri")
         val client = WebsocketClient()
-        client.connect(uri, origin)
+        try {
+            client.connect(uri, origin)
+        } catch (e: Exception) {
+            log.error("Failed to connect to websocket at $uri. Error: ${e.message}")
+            return
+        }
         client.onNewBlock {
             topic.onNext(it)
         }

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

@@ -1,14 +1,15 @@
 package io.emeraldpay.dshackle.upstream
 
+import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.grpc.Chain
-import io.infinitape.etherjar.rpc.json.BlockJson
-import io.infinitape.etherjar.rpc.json.TransactionJson
 import org.slf4j.LoggerFactory
+import java.util.concurrent.atomic.AtomicReference
 
 class Upstream(
         val chain: Chain,
         val api: EthereumApi,
-        private val ethereumWs: EthereumWs? = null
+        private val ethereumWs: EthereumWs? = null,
+        private val options: UpstreamsConfig.Options
 ) {
 
     private val log = LoggerFactory.getLogger(Upstream::class.java)
@@ -21,11 +22,23 @@ class Upstream(
         }
     }
 
+    val validator = UpstreamValidator(this, options)
+    private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE)
+
     init {
         log.info("Configured for ${chain.chainName}")
+
+        validator.start()
+                .subscribe {
+                    status.set(it)
+                }
     }
 
     fun isAvailable(): Boolean {
-        return true
+        return status.get() == UpstreamAvailability.OK
+    }
+
+    fun getStatus(): UpstreamAvailability {
+        return status.get()
     }
 }

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

@@ -0,0 +1,11 @@
+package io.emeraldpay.dshackle.upstream
+
+enum class UpstreamAvailability {
+
+    OK,
+    IMMATURE,
+    SYNCING,
+    LAGGING,
+    UNAVAILABLE
+
+}

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

@@ -0,0 +1,39 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.infinitape.etherjar.rpc.Batch
+import io.infinitape.etherjar.rpc.Commands
+import reactor.core.publisher.Flux
+import java.time.Duration
+import java.util.concurrent.TimeUnit
+
+class UpstreamValidator(
+        private val upstream: Upstream,
+        private val options: UpstreamsConfig.Options
+) {
+
+    fun validate(): UpstreamAvailability {
+        val batch = Batch()
+        val peerCount = batch.add(Commands.net().peerCount())
+        val syncing = batch.add(Commands.eth().syncing())
+        try {
+            upstream.api.execute(batch).get(5, TimeUnit.SECONDS)
+            if (syncing.get().isSyncing) {
+                return UpstreamAvailability.SYNCING
+            }
+            if (peerCount.get() < options.minPeers) {
+                return UpstreamAvailability.IMMATURE
+            }
+            return UpstreamAvailability.OK
+        } catch (e: Throwable) {
+            return UpstreamAvailability.UNAVAILABLE
+        }
+    }
+
+    fun start(): Flux<UpstreamAvailability> {
+        return Flux.interval(Duration.ofSeconds(15))
+                .map {
+                    validate()
+                }.onErrorContinue { _, _ -> UpstreamAvailability.UNAVAILABLE }
+    }
+}

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

@@ -1,7 +1,8 @@
 package io.emeraldpay.dshackle.upstream
 
 import com.fasterxml.jackson.databind.ObjectMapper
-import io.emeraldpay.dshackle.config.UpstreamsReader
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfigReader
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.rpc.DefaultRpcClient
 import io.infinitape.etherjar.rpc.transport.DefaultRpcTransport
@@ -9,6 +10,7 @@ import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.core.env.Environment
+import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Repository
 import java.io.File
 import java.net.URI
@@ -45,25 +47,40 @@ class Upstreams(
             System.exit(1)
         }
         log.info("Read upstream configuration from ${upstreamConfig.path}")
-        val reader = UpstreamsReader()
+        val reader = UpstreamsConfigReader()
         val config = reader.read(upstreamConfig.inputStream())
 
         val groups = HashMap<Chain, ArrayList<Upstream>>()
 
+        val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>()
+        config.defaultOptions.forEach { df ->
+            df.chains.forEach { chainName ->
+                chainNames[chainName]?.let {  chain ->
+                    var current = defaultOptions[chain]
+                    if (current == null) {
+                        current = df.options
+                    } else {
+                        current = current.merge(df.options)
+                    }
+                    defaultOptions[chain] = current
+                }
+            }
+        }
+
         config.upstreams.forEach { up ->
             val chain = chainNames[up.chain] ?: return@forEach
             var rpcApi: EthereumApi? = null
             var wsApi: EthereumWs? = null
             val urls = ArrayList<URI>()
             up.endpoints.forEach { endpoint ->
-                if (endpoint.type == io.emeraldpay.dshackle.config.Upstreams.EndpointType.JSON_RPC) {
+                if (endpoint.type == UpstreamsConfig.EndpointType.JSON_RPC) {
                     rpcApi = EthereumApi(
                             DefaultRpcClient(DefaultRpcTransport(endpoint.url)),
                             objectMapper,
                             chain
                     )
                 }
-                if (endpoint.type == io.emeraldpay.dshackle.config.Upstreams.EndpointType.WEBSOCKET) {
+                if (endpoint.type == UpstreamsConfig.EndpointType.WEBSOCKET) {
                     wsApi = EthereumWs(
                             endpoint.url,
                             endpoint.origin ?: URI("http://localhost")
@@ -72,10 +89,13 @@ class Upstreams(
                 }
                 urls.add(endpoint.url)
             }
+            val options = (up.options ?: UpstreamsConfig.Options())
+                    .merge(defaultOptions[chain])
+                    .merge(UpstreamsConfig.Options.getDefaults())
             if (rpcApi != null) {
                 log.info("Info using ${chain.chainName} upstream, at ${urls.joinToString()}")
                 val current = groups[chain] ?: ArrayList()
-                current.add(Upstream(chain, rpcApi!!, wsApi))
+                current.add(Upstream(chain, rpcApi!!, wsApi, options))
                 groups[chain] = current
             }
         }
@@ -87,4 +107,9 @@ class Upstreams(
     fun ethereumUpstream(chain: Chain): ChainConnect? {
         return chainMapping[chain]
     }
+
+    @Scheduled(fixedRate = 15000)
+    fun printStatuses() {
+        chainMapping.forEach { it.value.printStatus() }
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -2,9 +2,9 @@ package io.emeraldpay.dshackle.config
 
 import spock.lang.Specification
 
-class UpstreamsReaderSpec extends Specification {
+class UpstreamsConfigReaderSpec extends Specification {
 
-    UpstreamsReader reader = new UpstreamsReader()
+    UpstreamsConfigReader reader = new UpstreamsConfigReader()
 
     def "Parse standard config"() {
         setup:
@@ -30,11 +30,11 @@ class UpstreamsReaderSpec extends Specification {
             provider == "geth"
             endpoints.size() == 2
             with(endpoints.get(0)) {
-                type == Upstreams.EndpointType.JSON_RPC
+                type == UpstreamsConfig.EndpointType.JSON_RPC
                 url == new URI("http://localhost:8545")
             }
             with(endpoints.get(1)) {
-                type == Upstreams.EndpointType.WEBSOCKET
+                type == UpstreamsConfig.EndpointType.WEBSOCKET
                 url == new URI("ws://localhost:8546")
             }
         }
@@ -44,10 +44,10 @@ class UpstreamsReaderSpec extends Specification {
             provider == "infura"
             endpoints.size() == 1
             with(endpoints.get(0)) {
-                type == Upstreams.EndpointType.JSON_RPC
+                type == UpstreamsConfig.EndpointType.JSON_RPC
                 url == new URI("https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2")
-                auth instanceof Upstreams.BasicAuth
+                auth instanceof UpstreamsConfig.BasicAuth
-                with((Upstreams.BasicAuth)auth) {
+                with((UpstreamsConfig.BasicAuth)auth) {
                     key == "4fc258fe41a68149c199ad8f281f2015"
                 }
             }