solution: mark Upstream status right after WS connect/disconnect

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

@@ -61,7 +61,7 @@ abstract class DefaultUpstream(
         return status.get().status
     }
 
-    fun setStatus(avail: UpstreamAvailability) {
+    open fun setStatus(avail: UpstreamAvailability) {
         status.updateAndGet { curr ->
             Status(curr.lag, avail, statusByLag(curr.lag, avail))
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt

@@ -74,7 +74,8 @@ open class EthereumRpcUpstream(
 
     open fun createHead(): Head {
         return if (ethereumWsFactory != null) {
-            val ws = ethereumWsFactory.create(null).apply {
+            // do not set upstream to the WS, since it doesn't control the RPC upstream
+            val ws = ethereumWsFactory.create(null, null, null).apply {
                 connect()
             }
             val wsHead = EthereumWsHead(ws).apply {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt

@@ -32,7 +32,7 @@ import reactor.core.scheduler.Schedulers
 import java.time.Duration
 import java.util.concurrent.Executors
 
-class EthereumUpstreamValidator(
+open class EthereumUpstreamValidator(
         private val upstream: EthereumUpstream,
         private val options: UpstreamsConfig.Options
 ) {
@@ -43,7 +43,7 @@ class EthereumUpstreamValidator(
 
     private val objectMapper: ObjectMapper = Global.objectMapper
 
-    fun validate(): Mono<UpstreamAvailability> {
+    open fun validate(): Mono<UpstreamAvailability> {
         return upstream
                 .getApi()
                 .read(JsonRpcRequest("eth_syncing", listOf()))

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt

@@ -21,6 +21,8 @@ import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
@@ -64,15 +66,17 @@ class EthereumWsFactory(
 
     var basicAuth: AuthConfig.ClientBasicAuth? = null
 
-    fun create(rpcMetrics: RpcMetrics?): EthereumWs {
+    fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): EthereumWs {
-        return EthereumWs(uri, origin, basicAuth, rpcMetrics)
+        return EthereumWs(uri, origin, basicAuth, rpcMetrics, upstream, validator)
     }
 
     class EthereumWs(
             private val uri: URI,
             private val origin: URI,
             private val basicAuth: AuthConfig.ClientBasicAuth?,
-            private val rpcMetrics: RpcMetrics?
+            private val rpcMetrics: RpcMetrics?,
+            private val upstream: DefaultUpstream?,
+            private val validator: EthereumUpstreamValidator?
     ) : AutoCloseable {
 
         companion object {
@@ -154,6 +158,8 @@ class EthereumWsFactory(
             connection = HttpClient.create()
                     .doOnDisconnected {
                         log.info("Disconnected from $uri")
+                        // mark upstream as UNAVAIL
+                        upstream?.setStatus(UpstreamAvailability.UNAVAILABLE)
                         if (keepConnection) {
                             tryReconnectLater()
                         }
@@ -203,6 +209,9 @@ class EthereumWsFactory(
             //restart backoff after connection
             currentBackOff = reconnectBackoff.start()
 
+            //validate the connection, it can also be UNAVAIL if market as such after disconnect
+            validator?.validate()
+
             val consumer = inbound
                     // Accept up to 15Mb messages, same config is used by Geth
                     .aggregateFrames(15 * 1024 * 1024)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt

@@ -53,6 +53,7 @@ class EthereumWsUpstream(
     private val api: JsonRpcWsClient
 
     private var validatorSubscription: Disposable? = null
+    private val validator: EthereumUpstreamValidator
 
     init {
         val metricsTags = listOf(
@@ -72,7 +73,9 @@ class EthereumWsUpstream(
                         .register(Metrics.globalRegistry)
         )
 
-        connection = ethereumWsFactory.create(metrics)
+        validator = EthereumUpstreamValidator(this, getOptions())
+
+        connection = ethereumWsFactory.create(this, validator, metrics)
         head = EthereumWsHead(connection)
         api = JsonRpcWsClient(connection)
     }
@@ -102,7 +105,6 @@ class EthereumWsUpstream(
         head.start()
 
         log.debug("Start validation for upstream ${this.getId()}")
-        val validator = EthereumUpstreamValidator(this, getOptions())
         validatorSubscription = validator.start()
                 .subscribe(this::setStatus)
     }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy

@@ -1,6 +1,8 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.test.MockWSServer
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import reactor.test.StepVerifier
 import spock.lang.Shared
@@ -28,7 +30,7 @@ class EthereumWsFactoryRealSpec extends Specification {
         server = new MockWSServer(port)
         server.start()
         Thread.sleep(SLEEP)
-        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null)
+        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null, null)
     }
 
     def cleanup() {
@@ -91,6 +93,33 @@ class EthereumWsFactoryRealSpec extends Specification {
         act[0].value.contains("\"params\":[\"newHeads\"]")
     }
 
+    def "Gets UNAVAIL status right after disconnect"() {
+        setup:
+        def up = Mock(DefaultUpstream)
+        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null)
+        when:
+        conn.connect()
+        conn.reconnectIntervalSeconds = 10
+        Thread.sleep(SLEEP)
+        server.stop()
+        Thread.sleep(100)
+
+        then:
+        1 * up.setStatus(UpstreamAvailability.UNAVAILABLE)
+    }
+
+    def "Validates after connect"() {
+        setup:
+        def validator = Mock(EthereumUpstreamValidator)
+        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator, null)
+        when:
+        conn.connect()
+        Thread.sleep(100)
+
+        then:
+        1 * validator.validate()
+    }
+
     def "Try to connects to server until it's available"() {
         when:
         server.stop()

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy

@@ -53,7 +53,7 @@ class EthereumWsFactorySpec extends Specification {
 
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create()
+        def ws = wsf.create(null, null, null)
 
         apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block)
 
@@ -74,7 +74,7 @@ class EthereumWsFactorySpec extends Specification {
         def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create()
+        def ws = wsf.create(null, null, null)
 
         def tx = new TransactionJson().tap {
             hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
@@ -99,7 +99,7 @@ class EthereumWsFactorySpec extends Specification {
         def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create()
+        def ws = wsf.create(null, null, null)
 
         apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null)
 
@@ -122,7 +122,7 @@ class EthereumWsFactorySpec extends Specification {
         def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create()
+        def ws = wsf.create(null, null, null)
 
         apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"],
                 new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test"))