From 258507de4d90e9c3950703c2a8ef38264f51d56c Mon Sep 17 00:00:00 2001 From: msizov Date: Mon, 6 May 2024 15:15:28 +0700 Subject: [PATCH] Track web socket subscription_ids passed from grpc, retry eth_getLogs on errors/empty results (#465) * retry getLogs when result is empty or error * print error logs with grpc subscription ids * add logs on NativeSubscription's flux --- emerald-grpc | 2 +- .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 30 +++++++---- .../dshackle/rpc/NativeSubscribe.kt | 50 +++++++++++++------ .../ethereum/subscribe/ProduceLogs.kt | 27 ++++++++-- 4 files changed, 78 insertions(+), 31 deletions(-) diff --git a/emerald-grpc b/emerald-grpc index 708f2e8c..74434097 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 708f2e8c4a988c0cdfef3da9f281d5ced35ec7d9 +Subproject commit 744340971e13d081548788cf5e2e42ff291b1908 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 32e41bf1..601c7b07 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -108,15 +108,27 @@ class BlockchainRpc( override fun nativeSubscribe(request: Mono): Flux { var metrics: RequestMetrics? = null - return nativeSubscribe.nativeSubscribe( - request - .doOnNext { - metrics = chainMetrics.get(it.chain) - metrics!!.nativeSubscribeMetric.increment() - }, - ).doOnNext { - metrics?.nativeSubscribeRespMetric?.increment() - }.doOnError { failMetric.increment() } + return request + .doOnNext { + log.info("Starting subscription " + it.subscriptionId) + metrics = chainMetrics.get(it.chain) + metrics!!.nativeSubscribeMetric.increment() + }.flatMapMany { + req -> + nativeSubscribe.nativeSubscribe( + Mono.just(req), + ).doOnNext { + metrics?.nativeSubscribeRespMetric?.increment() + }.doOnComplete { + log.info("Subscription ${req.subscriptionId} completed") + }.doOnCancel { + log.info("Subscription ${req.subscriptionId} canceled") + }.doOnError { + t -> + log.info("Error ${t.message} in subscription ${req.subscriptionId}") + failMetric.increment() + } + } } override fun subscribeHead(request: Mono): Flux { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index ee024bbd..b8947522 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -46,20 +46,33 @@ open class NativeSubscribe( } private val objectMapper = Global.objectMapper - fun nativeSubscribe(request: Mono): Flux { return request - .flatMapMany(this@NativeSubscribe::start) - .map(this@NativeSubscribe::convertToProto) - .onErrorMap(this@NativeSubscribe::convertToStatus) + .flatMapMany { + it -> + val subscriptionId = it.subscriptionId + Mono.just(it) + .flatMapMany { + start(it, subscriptionId) + } + .map(this@NativeSubscribe::convertToProto) + .doOnCancel { + log.warn("Subscription $subscriptionId cancelled") + }.onErrorMap { + convertToStatus(it, subscriptionId) + } + } } - fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher { + fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher = start(request, "") + + fun start(request: BlockchainOuterClass.NativeSubscribeRequest, subscriptionId: String): Publisher { val chain = Chain.byId(request.chainValue) val multistream = getUpstream(chain) if (!multistream.getSubscriptionTopics().contains(request.method)) { + log.error("sub_id:" + subscriptionId + "subscribe ${request.method} is not supported for ${chain.chainCode}") return Mono.error(UnsupportedOperationException("subscribe ${request.method} is not supported for ${chain.chainCode}")) } @@ -81,33 +94,38 @@ open class NativeSubscribe( objectMapper.readValue(it.newInput(), List::class.java) } } - subscribe(chain, method, params, matcher) + subscribe(chain, method, params, matcher, subscriptionId) } return publisher.map { ResponseHolder(it, nonce) } } - fun convertToStatus(t: Throwable) = when (t) { - is SilentException.UnsupportedBlockchain -> StatusException( - Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}"), - ) + fun convertToStatus(t: Throwable) = convertToStatus(t, "") + fun convertToStatus(t: Throwable, subscriptionId: String = "") = when (t) { + is SilentException.UnsupportedBlockchain -> { + log.error("sub_id:$subscriptionId BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}") + StatusException(Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}")) + } - is UnsupportedOperationException -> StatusException( - Status.UNIMPLEMENTED.withDescription(t.message), - ) + is UnsupportedOperationException -> { + log.error("sub_id:$subscriptionId unimplemented error ${t.message}") + StatusException(Status.UNIMPLEMENTED.withDescription(t.message)) + } else -> { - log.warn("Unhandled error", t) + log.warn("sub_id:$subscriptionId Unhandled error", t) StatusException( Status.INTERNAL.withDescription(t.message), ) } } - open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux = + subscribe(chain, method, params, matcher) + + open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher, subscriptionId: String): Flux = getUpstream(chain).getEgressSubscription() .subscribe(method, params, matcher) .doOnError { - log.error("Error during subscription to $method, chain $chain, params $params", it) + log.error("sub_id:$subscriptionId Error during subscription to $method, chain $chain, params $params", it) } private fun getUpstream(chain: Chain): Multistream = diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt index be33befa..d998cd7a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result import io.emeraldpay.dshackle.upstream.ethereum.hex.HexData import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionLogJson @@ -28,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.kotlin.core.publisher.onErrorResume import reactor.kotlin.core.publisher.switchIfEmpty import java.util.concurrent.TimeUnit @@ -35,7 +37,7 @@ class ProduceLogs( private val logs: Reader>>, private val chain: Chain, ) { - + private val MAX_RETRIES = 3 companion object { private val log = LoggerFactory.getLogger(ProduceLogs::class.java) } @@ -73,11 +75,26 @@ class ProduceLogs( .map { it.copy(removed = true) } } - fun produceAdded(update: ConnectBlockUpdates.Update): Flux { + private fun produceAddedFallback(update: ConnectBlockUpdates.Update, retries: Int): Mono>> { return logs.read(update.blockHash).switchIfEmpty { - log.warn("Cannot find receipt for block ${update.blockHash} for chain ${chain.chainName}") - Mono.empty() - }.map { + if (retries > MAX_RETRIES) { + log.warn("Cannot find receipt for block ${update.blockHash} for chain ${chain.chainName} retries so far: $retries") + Mono.empty() + } else { + produceAddedFallback(update, retries + 1) + } + }.onErrorResume { t -> + if (retries > MAX_RETRIES) { + log.error("Error ${t.message} produced ${update.blockHash} for chain ${chain.chainName} retries so far: $retries") + Mono.empty() + } else { + produceAddedFallback(update, retries + 1) + } + } + } + + fun produceAdded(update: ConnectBlockUpdates.Update): Flux { + return produceAddedFallback(update, 0).map { it.data }.flatMapMany { val messages = it.map { log ->