diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt index 8b12b68c..aeb9dae2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt +++ b/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) 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 bfa5f742..761aaedf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt @@ -33,52 +33,29 @@ class SolanaLowerBoundBlockDetector( slot } } - .flatMap { slot -> - val from = if (slot <= 10) { - 1 - } else { - slot - 10 - } + .flatMap { reader.read( JsonRpcRequest( - "getBlocks", + "getBlock", // since getFirstAvailableBlock returns the slot of the lowest confirmed block we can directly call getBlock listOf( - from, - slot, + it, + mapOf( + "showRewards" to false, + "transactionDetails" to "none", + "maxSupportedTransactionVersion" to 0, + ), ), ), ) - } - .flatMap(JsonRpcResponse::requireResult) - .flatMap { - val response = Global.objectMapper.readValue(it, LongArray::class.java) - 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) - } - } + .flatMap(JsonRpcResponse::requireResult) + .map { blockData -> + val block = Global.objectMapper.readValue(blockData, SolanaBlock::class.java) + LowerBlockData(max(block.height, 1), it) + } } .retryWhen( Retry - .backoff(Long.MAX_VALUE, Duration.ofSeconds(1)) + .backoff(20, Duration.ofSeconds(1)) .maxBackoff(Duration.ofMinutes(3)) .doAfterRetry { log.debug( diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt index 64a60b91..1b6e4b18 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt +++ b/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 { diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt index c4c4969d..49605f6a 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt +++ b/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 { 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) } }