Disable futher finalized/safe requests for detectors if received "tag not supported" error

This commit is contained in:
msizov
2024-07-17 17:14:09 +05:00
committed by GitHub
parent 2e2a538b92
commit f1c669f96e
4 changed files with 43 additions and 16 deletions

View File

@@ -1,7 +1,9 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainRequest
import io.emeraldpay.dshackle.upstream.ChainResponse
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionRefJson import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionRefJson
@@ -13,6 +15,7 @@ import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.Sinks import reactor.core.publisher.Sinks
import java.time.Duration import java.time.Duration
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
@@ -23,12 +26,13 @@ class EthereumFinalizationDetector : FinalizationDetector {
} }
val data: ConcurrentHashMap<FinalizationType, FinalizationData> = ConcurrentHashMap() val data: ConcurrentHashMap<FinalizationType, FinalizationData> = ConcurrentHashMap()
private val disableDetector: ConcurrentHashMap<FinalizationType, Boolean> = ConcurrentHashMap()
private val finalizationSink = Sinks.many().multicast().directBestEffort<FinalizationData>() private val finalizationSink = Sinks.many().multicast().directBestEffort<FinalizationData>()
override fun detectFinalization( override fun detectFinalization(
upstream: Upstream, upstream: Upstream,
blockTime: Duration, blockTime: Duration,
chain: Chain,
): Flux<FinalizationData> { ): Flux<FinalizationData> {
return Flux.merge( return Flux.merge(
finalizationSink.asFlux(), finalizationSink.asFlux(),
@@ -56,23 +60,38 @@ class EthereumFinalizationDetector : FinalizationDetector {
), ),
), ),
).flatMap { (type, req) -> ).flatMap { (type, req) ->
upstream if (!disableDetector.getOrDefault(type, false)) {
.getIngressReader() upstream
.read(req) .getIngressReader()
.flatMap { .read(req)
it.requireResult().map { result -> .onErrorResume {
val block = if (it.message != null && it.message!!.matches(Regex("(bad request|block not found|Unknown block|tag not supported on pre-merge network)"))) {
Global.objectMapper log.warn("Can't retrieve tagged block, finalization detector for upstream ${upstream.getId()} $chain tag $type disabled")
.readValue(result, BlockJson::class.java) as BlockJson<TransactionRefJson>? disableDetector[type] = true
if (block != null) {
FinalizationData(block.number, type)
} else { } else {
throw RpcException(RpcResponseError.CODE_INVALID_JSON, "can't parse block data") throw it
}
Mono.empty<ChainResponse>()
}
.flatMap {
it.requireResult().map { result ->
val block = Global.objectMapper
.readValue(
result,
BlockJson::class.java,
) as BlockJson<TransactionRefJson>?
if (block != null) {
FinalizationData(block.number, type)
} else {
throw RpcException(RpcResponseError.CODE_INVALID_JSON, "can't parse block data")
}
} }
} }
} } else {
Flux.empty()
}
}.onErrorResume { }.onErrorResume {
log.error("Error during retrieving$it") log.error("Error in FinalizationDetector for upstream ${upstream.getId()} $chain$it")
Flux.empty() Flux.empty()
} }
}.filter { }.filter {

View File

@@ -1,5 +1,6 @@
package io.emeraldpay.dshackle.upstream.finalization package io.emeraldpay.dshackle.upstream.finalization
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import java.time.Duration import java.time.Duration
@@ -8,6 +9,7 @@ interface FinalizationDetector {
fun detectFinalization( fun detectFinalization(
upstream: Upstream, upstream: Upstream,
blockTime: Duration, blockTime: Duration,
chain: Chain,
): Flux<FinalizationData> ): Flux<FinalizationData>
fun getFinalizations(): Collection<FinalizationData> fun getFinalizations(): Collection<FinalizationData>
@@ -19,6 +21,7 @@ class NoopFinalizationDetector : FinalizationDetector {
override fun detectFinalization( override fun detectFinalization(
upstream: Upstream, upstream: Upstream,
blockTime: Duration, blockTime: Duration,
chain: Chain,
): Flux<FinalizationData> { ): Flux<FinalizationData> {
return Flux.empty() return Flux.empty()
} }

View File

@@ -308,7 +308,7 @@ open class GenericUpstream(
private fun detectFinalization() { private fun detectFinalization() {
finalizationDetectorSubscription = finalizationDetectorSubscription =
finalizationDetector.detectFinalization(this, chainConfig.expectedBlockTime).subscribe { finalizationDetector.detectFinalization(this, chainConfig.expectedBlockTime, chain).subscribe {
sendUpstreamStateEvent(UPDATED) sendUpstreamStateEvent(UPDATED)
} }
} }

View File

@@ -1,5 +1,6 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.reader.ChainReader
import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainRequest
@@ -33,9 +34,13 @@ class EthereumFinalizationDetectorTest {
val upstream = mock<Upstream> { val upstream = mock<Upstream> {
on { getIngressReader() } doReturn reader on { getIngressReader() } doReturn reader
} }
val chain = mock<Chain> {
on { toString() } doReturn "TestChain"
}
val detector = EthereumFinalizationDetector() val detector = EthereumFinalizationDetector()
StepVerifier.withVirtualTime { detector.detectFinalization(upstream, Duration.ofMillis(200)) } StepVerifier.withVirtualTime { detector.detectFinalization(upstream, Duration.ofMillis(200), chain) }
.expectSubscription() .expectSubscription()
.thenAwait(Duration.ofSeconds(0)) .thenAwait(Duration.ofSeconds(0))
.expectNext(FinalizationData(1L, FinalizationType.SAFE_BLOCK)) .expectNext(FinalizationData(1L, FinalizationType.SAFE_BLOCK))