Fixes lower data calculation (#410)

src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt

@@ -6,6 +6,7 @@ import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import java.time.Duration
 import java.time.Instant
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
 
 typealias LowerBoundBlockDetectorBuilder = (Chain, Upstream) -> LowerBoundBlockDetector
@@ -21,11 +22,20 @@ abstract class LowerBoundBlockDetector(
     protected val log = LoggerFactory.getLogger(this::class.java)
 
     fun lowerBlock(): Flux<LowerBlockData> {
+        val notProcessing = AtomicBoolean(true)
+
         return Flux.interval(
             Duration.ofSeconds(15),
-            Duration.ofSeconds(60),
+            Duration.ofMinutes(periodRequest()),
         )
-            .flatMap { lowerBlockDetect() }
+            .filter { notProcessing.get() }
+            .flatMap {
+                notProcessing.set(false)
+                lowerBlockDetect()
+            }
+            .doOnNext {
+                notProcessing.set(true)
+            }
             .filter { it.blockNumber > currentLowerBlock.get().blockNumber }
             .map {
                 log.info("Lower block of ${upstream.getId()} $chain: block height - {}, slot - {}", it.blockNumber, it.slot ?: "NA")
@@ -39,6 +49,8 @@ abstract class LowerBoundBlockDetector(
 
     protected abstract fun lowerBlockDetect(): Mono<LowerBlockData>
 
+    protected abstract fun periodRequest(): Long
+
     data class LowerBlockData(
         val blockNumber: Long,
         val slot: Long?,

src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt

@@ -27,7 +27,7 @@ abstract class RecursiveLowerBoundBlockDetector(
                 } else {
                     val middle = middleBlock(data)
 
-                    if (data.left > data.right) {
+                    if (data.left > data.right || middle == 0L) {
                         val current = if (data.current == 0L) 1 else data.current
                         Mono.just(LowerBoundData(current, true))
                     } else {
@@ -57,7 +57,7 @@ abstract class RecursiveLowerBoundBlockDetector(
             Long.MAX_VALUE,
             Duration.ofSeconds(1),
         )
-            .maxBackoff(Duration.ofSeconds(3))
+            .maxBackoff(Duration.ofMinutes(3))
             .filter {
                 !nonRetryableErrors.any { err -> it.message?.contains(err, true) ?: false }
             }
@@ -71,5 +71,9 @@ abstract class RecursiveLowerBoundBlockDetector(
             }
     }
 
+    override fun periodRequest(): Long {
+        return 10
+    }
+
     protected abstract fun hasState(blockNumber: Long): Mono<Boolean>
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt

@@ -34,6 +34,8 @@ class EthereumLowerBoundBlockDetector(
             "after last accepted block",
             "Version has either been pruned, or is for a future block", // cronos
             "no historical RPC is available for this historical", // optimism
+            "historical backend error", // optimism
+            "load state tree: failed to load state tree", // filecoin
         )
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt

@@ -79,7 +79,7 @@ class SolanaLowerBoundBlockDetector(
             .retryWhen(
                 Retry
                     .backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
-                    .maxBackoff(Duration.ofSeconds(3))
+                    .maxBackoff(Duration.ofMinutes(3))
                     .doAfterRetry {
                         log.debug(
                             "Error in calculation of lower block of upstream {}, retry attempt - {}, message - {}",
@@ -90,4 +90,8 @@ class SolanaLowerBoundBlockDetector(
                     },
             )
     }
+
+    override fun periodRequest(): Long {
+        return 3
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetector.kt

@@ -14,4 +14,8 @@ class StarknetLowerBoundBlockDetector(
     override fun lowerBlockDetect(): Mono<LowerBlockData> {
         return Mono.just(LowerBlockData(1))
     }
+
+    override fun periodRequest(): Long {
+        return 120
+    }
 }

src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt

@@ -38,7 +38,7 @@ class RecursiveLowerBoundBlockDetectorTest {
         StepVerifier.withVirtualTime { detector.lowerBlock() }
             .expectSubscription()
             .expectNoEvent(Duration.ofSeconds(15))
-            .expectNext(LowerBoundBlockDetector.LowerBlockData(17964844L))
+            .expectNextMatches { it.blockNumber == 17964844L }
             .thenCancel()
             .verify(Duration.ofSeconds(3))
 
@@ -64,7 +64,7 @@ class RecursiveLowerBoundBlockDetectorTest {
         StepVerifier.withVirtualTime { detector.lowerBlock() }
             .expectSubscription()
             .expectNoEvent(Duration.ofSeconds(15))
-            .expectNext(LowerBoundBlockDetector.LowerBlockData(1))
+            .expectNextMatches { it.blockNumber == 1L }
             .thenCancel()
             .verify(Duration.ofSeconds(3))
 

src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt

@@ -61,7 +61,7 @@ class SolanaLowerBoundBlockDetectorTest {
         StepVerifier.withVirtualTime { detector.lowerBlock() }
             .expectSubscription()
             .expectNoEvent(Duration.ofSeconds(15))
-            .expectNext(LowerBoundBlockDetector.LowerBlockData(21000000, 23000010))
+            .expectNextMatches { it.blockNumber == 21000000L && it.slot == 23000010L }
             .thenCancel()
             .verify(Duration.ofSeconds(3))