problem: WS upstream stuck in validation

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

@@ -31,6 +31,7 @@ import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 import java.time.Duration
 import java.util.concurrent.Executors
+import java.util.concurrent.TimeoutException
 
 open class EthereumUpstreamValidator(
     private val upstream: EthereumUpstream,
@@ -50,7 +51,10 @@ open class EthereumUpstreamValidator(
             .read(JsonRpcRequest("eth_syncing", listOf()))
             .flatMap(JsonRpcResponse::requireResult)
             .map { objectMapper.readValue(it, SyncingJson::class.java) }
-            .timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Syncing")))
+            .timeout(
+                Defaults.timeoutInternal,
+                Mono.fromCallable { log.warn("No response for eth_syncing from ${upstream.getId()}") }
+                    .then(Mono.error(TimeoutException("Validation timeout for Syncing"))))
             .flatMap { value ->
                 if (value.isSyncing) {
                     Mono.just(UpstreamAvailability.SYNCING)
@@ -60,7 +64,11 @@ open class EthereumUpstreamValidator(
                         .read(JsonRpcRequest("net_peerCount", listOf()))
                         .flatMap(JsonRpcResponse::requireStringResult)
                         .map(Integer::decode)
-                        .timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Peers")))
+                        .timeout(
+                            Defaults.timeoutInternal,
+                            Mono.fromCallable { log.warn("No response for net_peerCount from ${upstream.getId()}") }
+                                .then(Mono.error(TimeoutException("Validation timeout for Peers")))
+                        )
                         .map { count ->
                             val minPeers = options.minPeers ?: 1
                             if (count < minPeers) {

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

@@ -56,6 +56,7 @@ import java.time.Duration
 import java.util.Base64
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 
@@ -174,7 +175,9 @@ class WsConnection(
                     // going to try to reconnect later
                     tryReconnectLater()
                 },
-                { _, _ -> }
+                { _, t ->
+                    log.warn("Failed to process response from $uri. Error: ${t.message}")
+                }
             )
             .headers { headers ->
                 headers.add(HttpHeaderNames.ORIGIN, origin)
@@ -206,22 +209,6 @@ class WsConnection(
     }
 
     fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> {
-        // validate the connection, it can also be UNAVAIL if it's marked as such after a disconnect
-        if (validator != null) {
-            return validator.validate()
-                .flatMap {
-                    if (it == UpstreamAvailability.OK) {
-                        Mono.from(handleValidated(inbound, outbound))
-                    } else {
-                        tryReconnectLater()
-                        Mono.empty<Void>()
-                    }
-                }
-        }
-        return handleValidated(inbound, outbound)
-    }
-
-    fun handleValidated(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> {
         // restart backoff after connection
         currentBackOff = reconnectBackoff.start()
 
@@ -250,9 +237,6 @@ class WsConnection(
                 Mono.empty()
             }
 
-        val start = Mono.just(START_REQUEST).map {
-            Unpooled.wrappedBuffer(it.toByteArray())
-        }
         val calls = rpcSend
             .asFlux()
             .map {
@@ -261,13 +245,41 @@ class WsConnection(
 
         return outbound.send(
             Flux.merge(
-                start,
+                startWhenValidated(),
                 calls.subscribeOn(Schedulers.boundedElastic()),
                 consumer.then(Mono.empty<ByteBuf>()).subscribeOn(Schedulers.boundedElastic())
             )
         )
     }
 
+    /**
+     * Starts subscriptions ('newHeads') when the upstream is fully validated. If upstream is invalid it breaks flow with an Error.
+     * I.e., the first requests are made from a Validator and when it returns OK the Connection continues with other stuff.
+     */
+    fun startWhenValidated(): Publisher<ByteBuf> {
+        val start = Mono.just(START_REQUEST).map {
+            Unpooled.wrappedBuffer(it.toByteArray())
+        }
+
+        return if (validator != null) {
+            validator.validate()
+                .timeout(
+                    Defaults.timeoutInternal,
+                    Mono.fromCallable { log.warn("Not received a validation result from $uri") }.then(Mono.error(TimeoutException()))
+                )
+                .flatMap {
+                    if (it == UpstreamAvailability.OK) {
+                        start
+                    } else {
+                        tryReconnectLater()
+                        Mono.error(IllegalStateException("Upstream $uri is not ready"))
+                    }
+                }
+        } else {
+            start
+        }
+    }
+
     fun onRpc(msg: ResponseWSParser.WsResponse): Mono<Void> {
         return if (msg.id.isNumber()) {
             val resp = JsonRpcResponse(
@@ -359,23 +371,7 @@ class WsConnection(
 
     fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono<JsonRpcResponse> {
         val expectedId = request.id.toLong()
-        return Mono.just(request)
+        val failResponse = JsonRpcResponse(
-            .flatMap {
-                Flux.from(rpcReceive.asFlux())
-                    .doOnSubscribe { sendRpc(request) }
-                    .filter { resp -> resp.id.asNumber() == expectedId }
-                    .take(Defaults.timeout)
-                    .take(1)
-                    .singleOrEmpty()
-                    .doOnNext {
-                        rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS)
-                    }
-                    .doOnError {
-                        rpcMetrics?.errors?.increment()
-                    }
-                    .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) }
-                    .defaultIfEmpty(
-                        JsonRpcResponse(
             null,
             JsonRpcError(
                 RpcResponseError.CODE_INTERNAL_ERROR,
@@ -383,8 +379,17 @@ class WsConnection(
             ),
             JsonRpcResponse.Id.from(originalId)
         )
-                    )
+
-            }
+        return Flux.from(rpcReceive.asFlux())
+            .doOnSubscribe { sendRpc(request) }
+            .filter { resp -> resp.id.asNumber() == expectedId }
+            .take(Defaults.timeout)
+            .take(1)
+            .singleOrEmpty()
+            .doOnNext { rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) }
+            .doOnError { rpcMetrics?.errors?.increment() }
+            .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) }
+            .defaultIfEmpty(failResponse)
     }
 
     fun getBlocksFlux(): Flux<BlockContainer> {