From d6c847b210472fcc919d96b65306bf3f29239177 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 20 Dec 2021 20:55:46 -0500 Subject: [PATCH] problem: WS upstream stuck in validation --- .../ethereum/EthereumUpstreamValidator.kt | 12 ++- .../upstream/ethereum/WsConnection.kt | 99 ++++++++++--------- 2 files changed, 62 insertions(+), 49 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt index f7f6bfcc..09284440 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/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) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index 465bcc7c..6631ee90 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/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 { - // 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() - } - } - } - return handleValidated(inbound, outbound) - } - - fun handleValidated(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher { // 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()).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 { + 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 { return if (msg.id.isNumber()) { val resp = JsonRpcResponse( @@ -359,32 +371,25 @@ class WsConnection( fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono { val expectedId = request.id.toLong() - return Mono.just(request) - .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, - "Response not received from WebSocket" - ), - JsonRpcResponse.Id.from(originalId) - ) - ) - } + val failResponse = JsonRpcResponse( + null, + JsonRpcError( + RpcResponseError.CODE_INTERNAL_ERROR, + "Response not received from WebSocket" + ), + 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 {