better erigon bugged snapshot validator (#684)

* better erigon bugged snapshot validator

* add new error for erigon

* add error patterns for recursive lower bound

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

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.*"),
+        )
+    }
+}

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

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

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()) {
+) : EthereumLowerBoundDetectorBase(upstream.getChain()) {
-    private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, stateErrors, lowerBounds)
+    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",
         )
     }
 

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

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<ValidateUpstreamSettingsResult> {
+        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 ->
+                performErigonBugCheck()
-                    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
-                        }
-                    }
-                }
             }
         }.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<BigInteger> =
+    private fun performErigonBugCheck(): Mono<ValidateUpstreamSettingsResult> {
-        upstream.getIngressReader()
+        return getLatestBlockNumber()
-            .read(ChainRequest("eth_blockNumber", ListParams()))
+            .flatMap { latestBlock ->
-            .retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
+                binarySearchForBug(BigInteger.ONE, latestBlock, 0)
-                log.warn(
-                    "error during blockNumber retrieving for {}, iteration {}, reason - {}",
-                    upstream.getId(),
-                    ctx.iteration(),
-                    ctx.exception().message,
-                )
             }
+            .timeout(Duration.ofSeconds(10))
+            .onErrorReturn(ValidateUpstreamSettingsResult.UPSTREAM_VALID)
+    }
+
+    private fun getLatestBlockNumber(): Mono<BigInteger> {
+        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<Boolean> {
+    private fun binarySearchForBug(start: BigInteger, end: BigInteger, depth: Int): Mono<ValidateUpstreamSettingsResult> {
-        val tag = "0x${blockNumber.toString(16)}"
+        if (start >= end || depth > 20) {
-        return upstream.getIngressReader()
+            return Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_VALID)
-            .read(ChainRequest("eth_getBalance", ListParams(ZERO_ADDRESS, tag)))
+        }
-            .retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
+
-                log.warn(
+        val mid = start.add(end).divide(BigInteger.valueOf(2))
-                    "error during balance retrieving for {}, block {}, iteration {}, reason - {}",
+
-                    upstream.getId(),
+        return testBlockForBug(mid)
-                    tag,
+            .flatMap { result ->
-                    ctx.iteration(),
+                when (result) {
-                    ctx.exception().message,
+                    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<TestResult> {
+        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,
     }
 }

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<String>,
     protected val lowerBounds: LowerBounds,
+    protected val nonRetryableErrorPatters: Set<Regex> = 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) {