diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt index 1a0ccc32..4160d929 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt @@ -17,7 +17,7 @@ abstract class LowerBoundBlockDetector( ) { private val currentLowerBlock = AtomicReference(LowerBlockData.default()) - private val log = LoggerFactory.getLogger(this::class.java) + protected val log = LoggerFactory.getLogger(this::class.java) fun lowerBlock(): Flux { return Flux.interval( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt index de2ec575..98f87d40 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt @@ -2,6 +2,9 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.Chain import reactor.core.publisher.Mono +import reactor.util.retry.Retry +import reactor.util.retry.RetryBackoffSpec +import java.time.Duration abstract class RecursiveLowerBoundBlockDetector( chain: Chain, @@ -49,5 +52,24 @@ abstract class RecursiveLowerBoundBlockDetector( private fun middleBlock(lowerBoundData: LowerBoundData): Long = lowerBoundData.left + (lowerBoundData.right - lowerBoundData.left) / 2 + protected fun retrySpec(nonRetryableErrors: Set): RetryBackoffSpec { + return Retry.backoff( + Long.MAX_VALUE, + Duration.ofSeconds(1), + ) + .maxBackoff(Duration.ofSeconds(3)) + .filter { + !nonRetryableErrors.any { err -> it.message?.contains(err, true) ?: false } + } + .doAfterRetry { + log.debug( + "Error in calculation of lower block of upstream {}, retry attempt - {}, message - {}", + upstream.getId(), + it.totalRetries(), + it.failure().message, + ) + } + } + protected abstract fun hasState(blockNumber: Long): Mono } 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 74e9de6a..5d5ecc38 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt @@ -13,6 +13,28 @@ class EthereumLowerBoundBlockDetector( private val upstream: Upstream, ) : RecursiveLowerBoundBlockDetector(chain, upstream) { + companion object { + private val nonRetryableErrors = setOf( + "No state available for block", // nethermind + "missing trie node", // geth + "header not found", // optimism, bsc, avalanche + "Node state is pruned", // kava + "is not available, lowest height is", // kava, cronos + "State already discarded for", // moonriver, moonbeam + "your node is running with state pruning", // fuse + "Max height of block allowed", // mumbai + "failed to compute tipset state", // filecoin-calibration + "bad tipset height", // filecoin-calibration + "body not found for block", + "request beyond head block", + "block not found", + "could not find block", + "unknown block", + "header for hash not found", + "after last accepted block", + ) + } + override fun hasState(blockNumber: Long): Mono { return upstream.getIngressReader().read( JsonRpcRequest( @@ -20,6 +42,7 @@ class EthereumLowerBoundBlockDetector( listOf("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", blockNumber.toHex()), ), ) + .retryWhen(retrySpec(nonRetryableErrors)) .flatMap(JsonRpcResponse::requireResult) .map { true } .onErrorReturn(false) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt index e4d8edb4..04b0fad1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt @@ -13,6 +13,12 @@ class PolkadotLowerBoundBlockDetector( private val upstream: Upstream, ) : RecursiveLowerBoundBlockDetector(chain, upstream) { + companion object { + private val nonRetryableErrors = setOf( + "State already discarded for", + ) + } + override fun hasState(blockNumber: Long): Mono { return upstream.getIngressReader().read( JsonRpcRequest( @@ -32,6 +38,7 @@ class PolkadotLowerBoundBlockDetector( ), ) } + .retryWhen(retrySpec(nonRetryableErrors)) .flatMap(JsonRpcResponse::requireResult) .map { true } .onErrorReturn(false) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt index cfee2909..84293285 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt @@ -7,10 +7,13 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import reactor.core.publisher.Mono +import reactor.util.retry.Retry +import java.time.Duration +import kotlin.math.max class SolanaLowerBoundBlockDetector( chain: Chain, - upstream: Upstream, + private val upstream: Upstream, ) : LowerBoundBlockDetector(chain, upstream) { private val reader = upstream.getIngressReader() @@ -23,14 +26,24 @@ class SolanaLowerBoundBlockDetector( } .flatMap(JsonRpcResponse::requireResult) .map { - String(it).toLong() + val slot = String(it).toLong() + if (slot == 0L) { + 1L + } else { + slot + } } .flatMap { slot -> + val from = if (slot <= 10) { + 1 + } else { + slot - 10 + } reader.read( JsonRpcRequest( "getBlocks", listOf( - slot - 10, + from, slot, ), ), @@ -59,14 +72,24 @@ class SolanaLowerBoundBlockDetector( .flatMap(JsonRpcResponse::requireResult) .map { blockData -> val block = Global.objectMapper.readValue(blockData, SolanaBlock::class.java) - LowerBlockData(block.height, maxSlot) + LowerBlockData(max(block.height, 1), maxSlot) }.onErrorResume { Mono.empty() } } } - .onErrorResume { - Mono.empty() - } + .retryWhen( + Retry + .backoff(Long.MAX_VALUE, Duration.ofSeconds(1)) + .maxBackoff(Duration.ofSeconds(3)) + .doAfterRetry { + log.debug( + "Error in calculation of lower block of upstream {}, retry attempt - {}, message - {}", + upstream.getId(), + it.totalRetries(), + it.failure().message, + ) + }, + ) } } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt index 63a2e72f..486f3c77 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt @@ -92,7 +92,7 @@ class RecursiveLowerBoundBlockDetectorTest { } else { on { read(JsonRpcRequest("eth_getBalance", listOf("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", it.toHex()))) - } doReturn Mono.error(RuntimeException()) + } doReturn Mono.error(RuntimeException("missing trie node")) } } }, @@ -114,7 +114,7 @@ class RecursiveLowerBoundBlockDetectorTest { } doReturn Mono.just(JsonRpcResponse("\"$hash2\"".toByteArray(), null)) on { read(JsonRpcRequest("state_getMetadata", listOf(hash2))) - } doReturn Mono.error(RuntimeException()) + } doReturn Mono.error(RuntimeException("State already discarded for")) } } }, @@ -127,11 +127,8 @@ class RecursiveLowerBoundBlockDetectorTest { Arguments.of( mock { on { - read(JsonRpcRequest("chain_getBlockHash", listOf(any()))) - } doReturn Mono.just(JsonRpcResponse(ByteArray(0), null)) - on { - read(JsonRpcRequest("state_getMetadata", listOf(any()))) - } doReturn Mono.just(JsonRpcResponse(ByteArray(0), null)) + read(any()) + } doReturn Mono.just(JsonRpcResponse("\"0x1\"".toByteArray(), null)) }, PolkadotLowerBoundBlockDetector::class.java, ),