From 4b0650b8dc9d666b5c2a82cd95c2faa3b7c34a82 Mon Sep 17 00:00:00 2001 From: Vyacheslav Date: Wed, 16 Jul 2025 14:48:26 +0200 Subject: [PATCH] better erigon bugged snapshot validator (#684) * better erigon bugged snapshot validator * add new error for erigon * add error patterns for recursive lower bound --- .../EthereumLowerBoundBlockDetector.kt | 5 +- .../EthereumLowerBoundDetectorBase.kt | 13 ++ .../EthereumLowerBoundProofDetector.kt | 5 +- .../EthereumLowerBoundReceiptsDetector.kt | 5 +- .../EthereumLowerBoundStateDetector.kt | 8 +- .../ethereum/EthereumLowerBoundTxDetector.kt | 5 +- .../ethereum/EthereumUpstreamValidator.kt | 137 ++++++++++++------ .../detector/RecursiveLowerBound.kt | 4 +- 8 files changed, 118 insertions(+), 64 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundDetectorBase.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt index 24a8d69c..08e83d48 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt @@ -5,7 +5,6 @@ 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.lowerbound.toHex @@ -15,7 +14,7 @@ import reactor.core.publisher.Mono class EthereumLowerBoundBlockDetector( private val upstream: Upstream, -) : LowerBoundDetector(upstream.getChain()) { +) : EthereumLowerBoundDetectorBase(upstream.getChain()) { companion object { private const val NO_BLOCK_DATA = "No block data" @@ -31,7 +30,7 @@ class EthereumLowerBoundBlockDetector( ) } - private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, NO_BLOCK_ERRORS, lowerBounds) + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, NO_BLOCK_ERRORS, lowerBounds, commonErrorPatterns) override fun period(): Long { return 3 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundDetectorBase.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundDetectorBase.kt new file mode 100644 index 00000000..bee862d1 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundDetectorBase.kt @@ -0,0 +1,13 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector + +abstract class EthereumLowerBoundDetectorBase(chain: Chain) : LowerBoundDetector(chain) { + + companion object { + val commonErrorPatterns = setOf( + Regex(".*seekInFiles\\(invIndex=unknown index,txNum=\\d+\\) but data before txNum=\\d+ not available.*"), + ) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt index 67bf7c90..d24f36c1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt @@ -5,7 +5,6 @@ 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.lowerbound.toHex @@ -15,7 +14,7 @@ import reactor.core.publisher.Mono class EthereumLowerBoundProofDetector( private val upstream: Upstream, -) : LowerBoundDetector(upstream.getChain()) { +) : EthereumLowerBoundDetectorBase(upstream.getChain()) { companion object { private const val NO_PROOF_DATA = "distance to target block exceeds maximum proof window" @@ -31,7 +30,7 @@ class EthereumLowerBoundProofDetector( ) } - private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.PROOF, NO_PROOF_ERRORS, lowerBounds) + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.PROOF, NO_PROOF_ERRORS, lowerBounds, commonErrorPatterns) override fun period(): Long { return 3 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundReceiptsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundReceiptsDetector.kt index 0a3ace61..5856f070 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundReceiptsDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundReceiptsDetector.kt @@ -5,7 +5,6 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.ChainRequest 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.lowerbound.toHex @@ -14,7 +13,7 @@ import reactor.core.publisher.Flux class EthereumLowerBoundReceiptsDetector( private val upstream: Upstream, -) : LowerBoundDetector(upstream.getChain()) { +) : EthereumLowerBoundDetectorBase(upstream.getChain()) { companion object { const val MAX_OFFSET = 20 @@ -30,7 +29,7 @@ class EthereumLowerBoundReceiptsDetector( ).plus(EthereumLowerBoundBlockDetector.NO_BLOCK_ERRORS) } - private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.RECEIPTS, NO_RECEIPTS_ERRORS, lowerBounds) + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.RECEIPTS, NO_RECEIPTS_ERRORS, lowerBounds, commonErrorPatterns) override fun period(): Long { return 3 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt index 58fddcc9..835808e4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt @@ -6,7 +6,6 @@ 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.lowerbound.toHex @@ -16,8 +15,8 @@ import reactor.core.publisher.Mono class EthereumLowerBoundStateDetector( private val upstream: Upstream, -) : LowerBoundDetector(upstream.getChain()) { - private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, stateErrors, lowerBounds) +) : EthereumLowerBoundDetectorBase(upstream.getChain()) { + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, stateErrors, lowerBounds, commonErrorPatterns) companion object { val stateErrors = setOf( @@ -57,6 +56,9 @@ class EthereumLowerBoundStateDetector( "Access to archival, debug, or trace data is not included in your current plan", // chainstack "empty reader set", // strange bsc geth error "Request might be querying historical state that is not available", // monad + "No receipts data", + "No tx data", + "No block data", ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt index 03241f4e..f2a421e5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt @@ -5,7 +5,6 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.ChainRequest 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.lowerbound.toHex @@ -14,7 +13,7 @@ import reactor.core.publisher.Flux class EthereumLowerBoundTxDetector( private val upstream: Upstream, -) : LowerBoundDetector(upstream.getChain()) { +) : EthereumLowerBoundDetectorBase(upstream.getChain()) { companion object { const val MAX_OFFSET = 20 @@ -26,7 +25,7 @@ class EthereumLowerBoundTxDetector( ).plus(EthereumLowerBoundBlockDetector.NO_BLOCK_ERRORS) } - private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.TX, NO_TX_ERRORS, lowerBounds) + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.TX, NO_TX_ERRORS, lowerBounds, commonErrorPatterns) override fun period(): Long { return 3 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 d07836dd..e651d9f1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt @@ -234,36 +234,22 @@ class ErigonBuggedValidator( companion object { @JvmStatic val log: Logger = LoggerFactory.getLogger(ErigonBuggedValidator::class.java) - - private const val ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" - private val FIVE_THOUSAND = BigInteger.valueOf(5_000) } + private var callCount: Int = 0 + override fun validate(onError: ValidateUpstreamSettingsResult): Mono { + if (callCount % 10 != 0) { + callCount++ + return Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_VALID) + } + callCount++ + return isErigon().flatMap { isErigon -> if (!isErigon) { Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_VALID) } else { - latestBlockNumber().flatMap { latest -> - val pastBlock = latest.subtract(FIVE_THOUSAND).max(BigInteger.ZERO) - - Mono.zip( - balanceOkAt(pastBlock), // must succeed - balanceOkAt(BigInteger.ONE), // must fail - ).map { checks -> - val isBugged = checks.t1 && !checks.t2 - if (isBugged) { - log.warn( - "Erigon balance-bug detected on upstream {}: ok @ {}, error @ 0x1", - upstream.getId(), - pastBlock, - ) - ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR - } else { - ValidateUpstreamSettingsResult.UPSTREAM_VALID - } - } - } + performErigonBugCheck() } }.onErrorResume { ex -> log.error( @@ -290,35 +276,90 @@ class ErigonBuggedValidator( .map { it.lowercase().contains("erigon") } .doOnError { log.error("Error during execution 'web3_clientVersion' - {} for {}", it.message, upstream.getId()) } - private fun latestBlockNumber(): Mono = - upstream.getIngressReader() - .read(ChainRequest("eth_blockNumber", ListParams())) - .retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx -> - log.warn( - "error during blockNumber retrieving for {}, iteration {}, reason - {}", - upstream.getId(), - ctx.iteration(), - ctx.exception().message, - ) + private fun performErigonBugCheck(): Mono { + return getLatestBlockNumber() + .flatMap { latestBlock -> + binarySearchForBug(BigInteger.ONE, latestBlock, 0) } + .timeout(Duration.ofSeconds(10)) + .onErrorReturn(ValidateUpstreamSettingsResult.UPSTREAM_VALID) + } + + private fun getLatestBlockNumber(): Mono { + return upstream.getIngressReader() + .read(ChainRequest("eth_blockNumber", ListParams())) .flatMap(ChainResponse::requireStringResult) .map { BigInteger(it.removePrefix("0x"), 16) } - .doOnError { log.error("Error during execution 'eth_blockNumber' - {} for {}", it.message, upstream.getId()) } + .timeout(Duration.ofSeconds(1)) + .onErrorReturn(BigInteger.valueOf(1000000)) + } - private fun balanceOkAt(blockNumber: BigInteger): Mono { - val tag = "0x${blockNumber.toString(16)}" - return upstream.getIngressReader() - .read(ChainRequest("eth_getBalance", ListParams(ZERO_ADDRESS, tag))) - .retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx -> - log.warn( - "error during balance retrieving for {}, block {}, iteration {}, reason - {}", - upstream.getId(), - tag, - ctx.iteration(), - ctx.exception().message, - ) + private fun binarySearchForBug(start: BigInteger, end: BigInteger, depth: Int): Mono { + if (start >= end || depth > 20) { + return Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_VALID) + } + + val mid = start.add(end).divide(BigInteger.valueOf(2)) + + return testBlockForBug(mid) + .flatMap { result -> + when (result) { + TestResult.BUG_DETECTED -> { + log.warn( + "Erigon balance-bug detected on upstream {} at block {}", + upstream.getId(), + mid, + ) + Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR) + } + TestResult.VALID_RESPONSE -> { + // Go left (earlier blocks) + binarySearchForBug(start, mid, depth + 1) + } + TestResult.ERROR -> { + // Go right (later blocks) + binarySearchForBug(mid.add(BigInteger.ONE), end, depth + 1) + } + null -> Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_VALID) + } } - .flatMap { resp -> resp.requireStringResult().map { true } } - .onErrorResume { Mono.just(false) } + } + + private fun testBlockForBug(blockNumber: BigInteger): Mono { + val blockTag = "0x${blockNumber.toString(16)}" + val request = ChainRequest( + "eth_call", + ListParams( + mapOf( + "to" to "0x1111111111111111111111111111111111111111", + "data" to "0x1eaf190c", + ), + blockTag, + mapOf( + "0x1111111111111111111111111111111111111111" to mapOf( + "code" to "0x6080604052348015600e575f5ffd5b50600436106026575f3560e01c80631eaf190c14602a575b5f5ffd5b60306044565b604051603b91906078565b60405180910390f35b5f5f73ffffffffffffffffffffffffffffffffffffffff1631905090565b5f819050919050565b6072816062565b82525050565b5f60208201905060895f830184606b565b9291505056fea2646970667358221220251f5b4d2ed1abe77f66fde198a57ada08562dc3b0afbc6bac0261d1bf516b5d64736f6c634300081e0033", + ), + ), + ), + ) + + return upstream.getIngressReader() + .read(request) + .flatMap(ChainResponse::requireStringResult) + .map { result -> + when { + result == "0x" -> TestResult.BUG_DETECTED + result.startsWith("0x") && result.length > 2 -> TestResult.VALID_RESPONSE + else -> TestResult.ERROR + } + } + .timeout(Duration.ofSeconds(1)) + .onErrorReturn(TestResult.ERROR) + } + + private enum class TestResult { + BUG_DETECTED, + VALID_RESPONSE, + ERROR, } } 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 9debb77c..dcfcf0f0 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 @@ -19,6 +19,7 @@ open class RecursiveLowerBound( protected val type: LowerBoundType, protected val nonRetryableErrors: Set, protected val lowerBounds: LowerBounds, + protected val nonRetryableErrorPatters: Set = setOf(), ) { protected val log = LoggerFactory.getLogger(this::class.java) @@ -187,7 +188,8 @@ open class RecursiveLowerBound( ) .maxBackoff(Duration.ofMinutes(3)) .filter { - !nonRetryableErrors.any { err -> it.message?.contains(err, true) ?: false } + !nonRetryableErrors.any { err -> it.message?.contains(err, true) ?: false } && + !nonRetryableErrorPatters.any { err -> it.message?.matches(err) ?: false } } .doAfterRetry { if (it.totalRetries() > 30) {