Add reties for lower block calculation (#378)

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<LowerBlockData> {
         return Flux.interval(

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<String>): 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<Boolean>
 }

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<Boolean> {
         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)

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<Boolean> {
         return upstream.getIngressReader().read(
             JsonRpcRequest(
@@ -32,6 +38,7 @@ class PolkadotLowerBoundBlockDetector(
                     ),
                 )
             }
+            .retryWhen(retrySpec(nonRetryableErrors))
             .flatMap(JsonRpcResponse::requireResult)
             .map { true }
             .onErrorReturn(false)

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 {
+            .retryWhen(
-                Mono.empty()
+                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,
+                        )
+                    },
+            )
     }
 }

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<JsonRpcReader> {
                     on {
-                        read(JsonRpcRequest("chain_getBlockHash", listOf(any())))
+                        read(any())
-                    } doReturn Mono.just(JsonRpcResponse(ByteArray(0), null))
+                    } doReturn Mono.just(JsonRpcResponse("\"0x1\"".toByteArray(), null))
-                    on {
-                        read(JsonRpcRequest("state_getMetadata", listOf(any())))
-                    } doReturn Mono.just(JsonRpcResponse(ByteArray(0), null))
                 },
                 PolkadotLowerBoundBlockDetector::class.java,
             ),