Improve solana lower bound detection (#416)

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

@@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.Chain
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.kotlin.core.publisher.switchIfEmpty
 import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.atomic.AtomicBoolean
@@ -32,6 +33,8 @@ abstract class LowerBoundBlockDetector(
             .flatMap {
                 notProcessing.set(false)
                 lowerBlockDetect()
+                    .onErrorResume { Mono.just(LowerBlockData.default()) }
+                    .switchIfEmpty { Mono.just(LowerBlockData.default()) } // just to trigger onNext event
             }
             .doOnNext {
                 notProcessing.set(true)

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

@@ -33,52 +33,29 @@ class SolanaLowerBoundBlockDetector(
                     slot
                 }
             }
-            .flatMap { slot ->
+            .flatMap {
-                val from = if (slot <= 10) {
-                    1
-                } else {
-                    slot - 10
-                }
                 reader.read(
                     JsonRpcRequest(
-                        "getBlocks",
+                        "getBlock", // since getFirstAvailableBlock returns the slot of the lowest confirmed block we can directly call getBlock
                         listOf(
-                            from,
+                            it,
-                            slot,
+                            mapOf(
+                                "showRewards" to false,
+                                "transactionDetails" to "none",
+                                "maxSupportedTransactionVersion" to 0,
+                            ),
                         ),
                     ),
                 )
-            }
+                    .flatMap(JsonRpcResponse::requireResult)
-            .flatMap(JsonRpcResponse::requireResult)
+                    .map { blockData ->
-            .flatMap {
+                        val block = Global.objectMapper.readValue(blockData, SolanaBlock::class.java)
-                val response = Global.objectMapper.readValue(it, LongArray::class.java)
+                        LowerBlockData(max(block.height, 1), it)
-                if (response == null || response.isEmpty()) {
+                    }
-                    Mono.empty()
-                } else {
-                    val maxSlot = response.max()
-                    reader.read(
-                        JsonRpcRequest(
-                            "getBlock",
-                            listOf(
-                                maxSlot,
-                                mapOf(
-                                    "showRewards" to false,
-                                    "transactionDetails" to "none",
-                                    "maxSupportedTransactionVersion" to 0,
-                                ),
-                            ),
-                        ),
-                    )
-                        .flatMap(JsonRpcResponse::requireResult)
-                        .map { blockData ->
-                            val block = Global.objectMapper.readValue(blockData, SolanaBlock::class.java)
-                            LowerBlockData(max(block.height, 1), maxSlot)
-                        }
-                }
             }
             .retryWhen(
                 Retry
-                    .backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
+                    .backoff(20, Duration.ofSeconds(1))
                     .maxBackoff(Duration.ofMinutes(3))
                     .doAfterRetry {
                         log.debug(

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

@@ -6,7 +6,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundBlockDetector
 import io.emeraldpay.dshackle.upstream.polkadot.PolkadotLowerBoundBlockDetector
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.Arguments
 import org.junit.jupiter.params.provider.MethodSource
@@ -42,7 +42,7 @@ class RecursiveLowerBoundBlockDetectorTest {
             .thenCancel()
             .verify(Duration.ofSeconds(3))
 
-        Assertions.assertEquals(LowerBoundBlockDetector.LowerBlockData(17964844L), detector.getCurrentLowerBlock())
+        assertEquals(17964844L, detector.getCurrentLowerBlock().blockNumber)
     }
 
     @ParameterizedTest
@@ -68,7 +68,7 @@ class RecursiveLowerBoundBlockDetectorTest {
             .thenCancel()
             .verify(Duration.ofSeconds(3))
 
-        Assertions.assertEquals(LowerBoundBlockDetector.LowerBlockData(1), detector.getCurrentLowerBlock())
+        assertEquals(1, detector.getCurrentLowerBlock().blockNumber)
     }
 
     companion object {

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

@@ -3,11 +3,10 @@ package io.emeraldpay.dshackle.upstream.solana
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.reader.JsonRpcReader
-import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.mock
@@ -22,14 +21,12 @@ class SolanaLowerBoundBlockDetectorTest {
         val reader = mock<JsonRpcReader> {
             on { read(JsonRpcRequest("getFirstAvailableBlock", listOf())) } doReturn
                 Mono.just(JsonRpcResponse("25000000".toByteArray(), null))
-            on { read(JsonRpcRequest("getBlocks", listOf(24999990L, 25000000L))) } doReturn
-                Mono.just(JsonRpcResponse("[23000000, 23000005, 23000010]".toByteArray(), null))
             on {
                 read(
                     JsonRpcRequest(
                         "getBlock",
                         listOf(
-                            23000010L,
+                            25000000L,
                             mapOf(
                                 "showRewards" to false,
                                 "transactionDetails" to "none",
@@ -61,10 +58,11 @@ class SolanaLowerBoundBlockDetectorTest {
         StepVerifier.withVirtualTime { detector.lowerBlock() }
             .expectSubscription()
             .expectNoEvent(Duration.ofSeconds(15))
-            .expectNextMatches { it.blockNumber == 21000000L && it.slot == 23000010L }
+            .expectNextMatches { it.blockNumber == 21000000L && it.slot == 25000000L }
             .thenCancel()
             .verify(Duration.ofSeconds(3))
 
-        Assertions.assertEquals(LowerBoundBlockDetector.LowerBlockData(21000000, 23000010), detector.getCurrentLowerBlock())
+        assertEquals(21000000, detector.getCurrentLowerBlock().blockNumber)
+        assertEquals(25000000, detector.getCurrentLowerBlock().slot)
     }
 }