diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt index b23fd67e..85727e89 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt @@ -148,6 +148,8 @@ class ChainEventMapper { LowerBoundType.LOGS -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS LowerBoundType.TRACE -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TRACE LowerBoundType.PROOF -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF + LowerBoundType.BLOB -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB + LowerBoundType.EPOCH -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 6f7e833e..8d1a71ed 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -108,6 +108,8 @@ class StreamHead( LowerBoundType.LOGS -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS LowerBoundType.TRACE -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TRACE LowerBoundType.PROOF -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF + LowerBoundType.BLOB -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB + LowerBoundType.EPOCH -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlobDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlobDetector.kt new file mode 100644 index 00000000..6e83959d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlobDetector.kt @@ -0,0 +1,66 @@ +package io.emeraldpay.dshackle.upstream.beaconchain + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.module.kotlin.readValue +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.ChainCallError +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound +import io.emeraldpay.dshackle.upstream.rpcclient.RestParams +import reactor.core.publisher.Flux +import reactor.kotlin.core.publisher.toFlux + +class BeaconChainLowerBoundBlobDetector( + private val chain: Chain, + private val upstream: Upstream, +) : LowerBoundDetector(chain) { + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOB, stateErrors, lowerBounds) + + companion object { + const val MAX_OFFSET = 20 + val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404} + val notFoundError2 = "lock not found" + val notFoundError3 = "has not been found" + val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3) + } + + override fun period(): Long { + return 5 + } + + override fun internalDetectLowerBound(): Flux { + return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { block -> + val restParams = RestParams(emptyList(), emptyList(), listOf(block.toString()), ByteArray(0)) + + upstream.getIngressReader() + .read(ChainRequest("GET#/eth/v1/beacon/blob_sidecars/*", restParams)) + .flatMap(ChainResponse::requireResult) + .timeout(Defaults.internalCallsTimeout) + .map { + parseHeadersResponse(it) + } + }.toFlux() + } + + override fun types(): Set { + return setOf(LowerBoundType.BLOB) + } + + private fun parseHeadersResponse(data: ByteArray): ChainResponse { + val node = Global.objectMapper.readValue(data) + if (node.get("code") != null && node.get("message") != null && node.get("code").textValue() == "404") { + return ChainResponse(null, ChainCallError(node.get("code").asInt(), node.get("message").asText(), node.get("message").asText())) + } + if (node.get("data").toString() == "[]") { + return ChainResponse(null, ChainCallError(404, notFoundError)) + } + return ChainResponse(node.get("data").toString().toByteArray(), null) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt index 96d7cdcc..d44f2c5c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt @@ -24,11 +24,13 @@ class BeaconChainLowerBoundBlockDetector( private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, stateErrors, lowerBounds) companion object { - const val MAX_OFFSET = 30 + const val MAX_OFFSET = 20 val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404} val notFoundError2 = "Could not find requested block" // {"message":"Could not find requested block: signed beacon block can't be nil","code":404} val notFoundError3 = "has not been found" // Block header/data has not been found - val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3) + val notFoundError4 = "lock not found" // {"message":"block not found 1413","code":404} + val notFoundError5 = "Internal Server Error" // block pi returns {"message":"Internal Server Error"} in first 9 blocks (?) + val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3, notFoundError4, notFoundError5) } override fun period(): Long { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundEpochDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundEpochDetector.kt new file mode 100644 index 00000000..4be97020 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundEpochDetector.kt @@ -0,0 +1,96 @@ +package io.emeraldpay.dshackle.upstream.beaconchain + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.module.kotlin.readValue +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.ChainCallError +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBounds +import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound +import io.emeraldpay.dshackle.upstream.rpcclient.RestParams +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.kotlin.core.publisher.toFlux + +class EpochRecursiveLowerBound( + upstream: Upstream, + type: LowerBoundType, + nonRetryableErrors: Set, + lowerBounds: LowerBounds, +) : RecursiveLowerBound(upstream, type, nonRetryableErrors, lowerBounds) { + + // similar to recursive lower bound, but range is adjusted as epoch range is 32x smaller than height + override fun initialRange(): Mono { + return Mono.just(upstream.getHead()) + .flatMap { + val currentHeight = it.getCurrentHeight() + if (currentHeight == null) { + Mono.empty() + } else if (lowerBounds.getLastBound(type) == null) { + Mono.just(LowerBoundBinarySearchData(0, currentHeight / 32)) // 1 epoch is 32 slots + } else { + Mono.just(LowerBoundBinarySearchData(lowerBounds.getLastBound(type)!!.lowerBound, currentHeight / 32)) // 1 epoch is 32 slots + } + } + } +} + +class BeaconChainLowerBoundEpochDetector( + private val chain: Chain, + private val upstream: Upstream, +) : LowerBoundDetector(chain) { + private val recursiveLowerBound = EpochRecursiveLowerBound(upstream, LowerBoundType.EPOCH, stateErrors, lowerBounds) + + companion object { + const val MAX_OFFSET = 20 + val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404} + val notFoundError2 = "Could not get requested state" + val notFoundError3 = "missing state" // "missing state at slot 11609023" + val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3) + } + + override fun period(): Long { + return 5 + } + + override fun internalDetectLowerBound(): Flux { + return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { slot -> + val restParams = RestParams(listOf(), emptyList(), listOf(slot.toString()), "[\"1\"]".toByteArray()) + + upstream.getIngressReader() + .read(ChainRequest("POST#/eth/v1/beacon/rewards/attestations/*", restParams)) + .flatMap(ChainResponse::requireResult) + .timeout(Defaults.internalCallsTimeout) + .map { + parseHeadersResponse(it) + } + }.toFlux() + } + + override fun types(): Set { + return setOf(LowerBoundType.EPOCH) + } + + private fun parseHeadersResponse(data: ByteArray): ChainResponse { + val node = Global.objectMapper.readValue(data) + if (node.get("code") != null && node.get("message") != null && node.get("code").textValue() == "404") { + return ChainResponse(null, ChainCallError(node.get("code").asInt(), node.get("message").asText(), node.get("message").asText())) + } + + val jsonData = node.get("data") + if (jsonData != null) { + val str = jsonData.toString() + if (str.length >= 2) { + return ChainResponse(str.toByteArray(), null) + } + } + return ChainResponse(null, ChainCallError(404, notFoundError)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt index e1649532..e9d0fcf3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt @@ -10,6 +10,11 @@ class BeaconChainLowerBoundService( private val upstream: Upstream, ) : LowerBoundService(chain, upstream) { override fun detectors(): List { - return listOf(BeaconChainLowerBoundBlockDetector(chain, upstream)) + return listOf( + BeaconChainLowerBoundBlockDetector(chain, upstream), + BeaconChainLowerBoundEpochDetector(chain, upstream), + BeaconChainLowerBoundStateDetector(chain, upstream), + BeaconChainLowerBoundBlobDetector(chain, upstream), + ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt new file mode 100644 index 00000000..e84fdca5 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt @@ -0,0 +1,70 @@ +package io.emeraldpay.dshackle.upstream.beaconchain + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.module.kotlin.readValue +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.ChainCallError +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound +import io.emeraldpay.dshackle.upstream.rpcclient.RestParams +import reactor.core.publisher.Flux +import reactor.kotlin.core.publisher.toFlux + +class BeaconChainLowerBoundStateDetector( + private val chain: Chain, + private val upstream: Upstream, +) : LowerBoundDetector(chain) { + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, stateErrors, lowerBounds) + + companion object { + const val MAX_OFFSET = 20 + val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404} + val notFoundError2 = "Could not get requested state" + val stateErrors = setOf(notFoundError, notFoundError2) + } + + override fun period(): Long { + return 5 + } + + override fun internalDetectLowerBound(): Flux { + return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { slot -> + val restParams = RestParams(listOf(), emptyList(), listOf(slot.toString()), "[\"1\"]".toByteArray()) + + upstream.getIngressReader() + .read(ChainRequest("POST#/eth/v1/beacon/states/*/validator_balances", restParams)) + .flatMap(ChainResponse::requireResult) + .timeout(Defaults.internalCallsTimeout) + .map { + parseHeadersResponse(it) + } + }.toFlux() + } + + override fun types(): Set { + return setOf(LowerBoundType.STATE) + } + + private fun parseHeadersResponse(data: ByteArray): ChainResponse { + val node = Global.objectMapper.readValue(data) + if (node.get("code") != null && node.get("message") != null && node.get("code").textValue() == "404") { + return ChainResponse(null, ChainCallError(node.get("code").asInt(), node.get("message").asText(), node.get("message").asText())) + } + + val jsonData = node.get("data") + if (jsonData != null) { + val str = jsonData.toString() + if (str.length >= 2) { + return ChainResponse(str.toByteArray(), null) + } + } + return ChainResponse(null, ChainCallError(404, notFoundError)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt index c4ca05f2..333ff1a3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt @@ -20,7 +20,7 @@ data class LowerBoundData( } enum class LowerBoundType { - UNKNOWN, STATE, SLOT, BLOCK, TX, LOGS, TRACE, PROOF + UNKNOWN, STATE, SLOT, BLOCK, TX, LOGS, TRACE, PROOF, BLOB, EPOCH } fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType { @@ -34,7 +34,7 @@ fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType { BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS -> LowerBoundType.LOGS BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TRACE -> LowerBoundType.TRACE BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF -> LowerBoundType.PROOF - BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB -> LowerBoundType.UNKNOWN - BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH -> LowerBoundType.UNKNOWN + BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB -> LowerBoundType.BLOB + BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH -> LowerBoundType.EPOCH } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt index dce37336..9debb77c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt @@ -14,13 +14,13 @@ import reactor.util.retry.RetryBackoffSpec import java.time.Duration import java.util.concurrent.atomic.AtomicInteger -class RecursiveLowerBound( - private val upstream: Upstream, - private val type: LowerBoundType, - private val nonRetryableErrors: Set, - private val lowerBounds: LowerBounds, +open class RecursiveLowerBound( + protected val upstream: Upstream, + protected val type: LowerBoundType, + protected val nonRetryableErrors: Set, + protected val lowerBounds: LowerBounds, ) { - private val log = LoggerFactory.getLogger(this::class.java) + protected val log = LoggerFactory.getLogger(this::class.java) fun recursiveDetectLowerBound(hasData: (Long) -> Mono): Flux { return initialRange() @@ -32,7 +32,17 @@ class RecursiveLowerBound( if (data.left > data.right) { val current = if (data.current == 0L) 1 else data.current - Mono.just(LowerBoundBinarySearchData(current, true)) + hasData(current) + .retryWhen(retrySpec(middle, nonRetryableErrors)) + .flatMap(ChainResponse::requireResult) + .map { LowerBoundBinarySearchData(current, true) } + .onErrorResume { + if (current == 1L && data.right > 10) { + Mono.empty() // Couldn't detect bound: data.right is chain height here and current wasn't set because we got errors all the time + } else { + Mono.just(LowerBoundBinarySearchData(current, true)) // if we approached left bound(1) or node have just pruned data.current due to long bound calculation, return data.current as is + } + } } else { hasData(middle) .retryWhen(retrySpec(middle, nonRetryableErrors)) @@ -75,7 +85,17 @@ class RecursiveLowerBound( if (data.left > data.right) { val current = if (data.current == 0L) 1 else data.current - Mono.just(LowerBoundBinarySearchData(current, true)) + hasData(current) + .retryWhen(retrySpec(middle, nonRetryableErrors)) + .flatMap(ChainResponse::requireResult) + .map { LowerBoundBinarySearchData(current, true) } + .onErrorResume { + if (current == 1L && data.right > 10) { + Mono.empty() // Couldn't detect bound: data.right is chain height here and current wasn't set because we got errors all the time + } else { + Mono.just(LowerBoundBinarySearchData(current, true)) // if we approached left bound(1) or node have just pruned data.current due to long bound calculation, return data.current as is + } + } } else { hasData(middle) .retryWhen(retrySpec(middle, nonRetryableErrors)) @@ -99,7 +119,7 @@ class RecursiveLowerBound( ) } - private fun shiftLeftAndSearch( + protected fun shiftLeftAndSearch( currentData: LowerBoundBinarySearchData, currentMiddle: Long, visitedBlocks: HashSet, @@ -145,7 +165,7 @@ class RecursiveLowerBound( } } - private fun initialRange(): Mono { + protected open fun initialRange(): Mono { return Mono.just(upstream.getHead()) .flatMap { val currentHeight = it.getCurrentHeight() @@ -160,7 +180,7 @@ class RecursiveLowerBound( } } - private fun retrySpec(block: Long, nonRetryableErrors: Set): RetryBackoffSpec { + protected fun retrySpec(block: Long, nonRetryableErrors: Set): RetryBackoffSpec { return Retry.backoff( Long.MAX_VALUE, Duration.ofSeconds(1), @@ -172,8 +192,9 @@ class RecursiveLowerBound( .doAfterRetry { if (it.totalRetries() > 30) { log.warn( - "There are too much retries to calculate {} lower bound of upstream {}, " + + "There are too much retries to calculate {} lower bound of upstream {}, block {} " + "probably this error with message `{}` is not retryable, please report it to dshackle devs", + block, type, upstream.getId(), it.failure().message, @@ -191,10 +212,10 @@ class RecursiveLowerBound( } } - private fun middleBlock(lowerBoundBinarySearchData: LowerBoundBinarySearchData): Long = + protected fun middleBlock(lowerBoundBinarySearchData: LowerBoundBinarySearchData): Long = lowerBoundBinarySearchData.left + (lowerBoundBinarySearchData.right - lowerBoundBinarySearchData.left) / 2 - private data class LowerBoundBinarySearchData( + protected data class LowerBoundBinarySearchData( val left: Long, val right: Long, val current: Long,