diff --git a/emerald-grpc b/emerald-grpc index 2ed74d52..5b3b521c 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 2ed74d52c8585d1c74da5e2a78d4b7234dd70ecb +Subproject commit 5b3b521ca5dad7eaa2fb888910f49ab6853dfb15 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index a3af821d..4fafa133 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -95,6 +95,7 @@ class StreamHead( LowerBoundType.SLOT -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_SLOT LowerBoundType.UNKNOWN -> BlockchainOuterClass.LowerBoundType.UNRECOGNIZED LowerBoundType.STATE -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_STATE + LowerBoundType.BLOCK -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOCK } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt new file mode 100644 index 00000000..a58c2c11 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt @@ -0,0 +1,44 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound +import io.emeraldpay.dshackle.upstream.lowerbound.toHex +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +class EthereumLowerBoundBlockDetector( + private val upstream: Upstream, +) : LowerBoundDetector() { + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, setOf("No block data")) + + override fun period(): Long { + return 5 + } + + override fun internalDetectLowerBound(): Flux { + return recursiveLowerBound.recursiveDetectLowerBound { block -> + if (block == 0L) { + Mono.just(ChainResponse(ByteArray(0), null)) + } else { + upstream.getIngressReader() + .read( + ChainRequest( + "eth_getBlockByNumber", + ListParams(block.toHex(), false), + ), + ) + .doOnNext { + if (it.hasResult() && it.getResult().contentEquals("null".toByteArray())) { + throw IllegalStateException("No block data") + } + } + } + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt index eb01605e..1211a5b0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt @@ -10,6 +10,9 @@ class EthereumLowerBoundService( private val upstream: Upstream, ) : LowerBoundService(chain, upstream) { override fun detectors(): List { - return listOf(EthereumLowerBoundStateDetector(upstream)) + return listOf( + EthereumLowerBoundStateDetector(upstream), + EthereumLowerBoundBlockDetector(upstream), + ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt index d0c5aa65..47342b66 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt @@ -19,5 +19,5 @@ data class LowerBoundData( } enum class LowerBoundType { - UNKNOWN, STATE, SLOT + UNKNOWN, STATE, SLOT, BLOCK } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt index 7aa18f12..6a4bd722 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt @@ -11,6 +11,7 @@ import io.emeraldpay.dshackle.upstream.lowerbound.toHex import io.emeraldpay.dshackle.upstream.polkadot.PolkadotLowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.Arguments import org.junit.jupiter.params.provider.MethodSource @@ -22,22 +23,93 @@ import reactor.test.StepVerifier import java.time.Duration class RecursiveLowerBoundServiceTest { + private val blocks = listOf( + 9000000L, 13500000L, 15750000L, 16875000L, 17437500L, 17718750L, 17859375L, 17929688L, 17964844L, 17947266L, + 17956055L, 17960449L, 17962646L, 17963745L, 17964294L, 17964569L, 17964706L, 17964775L, 17964809L, 17964826L, + 17964835L, 17964839L, 17964841L, 17964842L, 17964843L, + ) + private val hash1 = "0x1b1a5dd69e12aa12e2b9197be0d0cceef3dde6368ea6376ad7c8b06488c9cf6a" + private val hash2 = "0x1b1a5dd69e12aa12e2b9197be0d0cceef3dde6368ea6376ad7c8b06488c9cf7a" - @ParameterizedTest - @MethodSource("detectors") - fun `find lower block closer to the height`( - reader: ChainReader, - detectorClass: Class, - ) { + @Test + fun `find lower data for eth`() { val head = mock { on { getCurrentHeight() } doReturn 18000000 } + val reader = mock { + blocks.forEach { + if (it == 17964844L) { + on { + read(ChainRequest("eth_getBalance", ListParams(ZERO_ADDRESS, it.toHex()))) + } doReturn Mono.just(ChainResponse(ByteArray(0), null)) + on { + read(ChainRequest("eth_getBlockByNumber", ListParams(it.toHex(), false))) + } doReturn Mono.just(ChainResponse(ByteArray(0), null)) + } else { + on { + read(ChainRequest("eth_getBalance", ListParams(ZERO_ADDRESS, it.toHex()))) + } doReturn Mono.error(RuntimeException("missing trie node")) + on { + read(ChainRequest("eth_getBlockByNumber", ListParams(it.toHex(), false))) + } doReturn Mono.error(RuntimeException("No block data")) + } + } + } val upstream = mock { on { getHead() } doReturn head on { getIngressReader() } doReturn reader } - val detector = detectorClass.getConstructor(Chain::class.java, Upstream::class.java).newInstance(Chain.UNSPECIFIED, upstream) + val detector = EthereumLowerBoundService(Chain.UNSPECIFIED, upstream) + + StepVerifier.withVirtualTime { detector.detectLowerBounds() } + .expectSubscription() + .expectNoEvent(Duration.ofSeconds(15)) + .expectNextMatches { it.lowerBound == 17964844L && it.type == LowerBoundType.STATE } + .expectNextMatches { it.lowerBound == 17964844L && it.type == LowerBoundType.BLOCK } + .thenCancel() + .verify(Duration.ofSeconds(3)) + + assertThat(detector.getLowerBounds().toList()) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields("timestamp") + .hasSameElementsAs( + listOf( + LowerBoundData(17964844L, LowerBoundType.STATE), + LowerBoundData(17964844L, LowerBoundType.BLOCK), + ), + ) + } + + @Test + fun `find lower data for polka`() { + val head = mock { + on { getCurrentHeight() } doReturn 18000000 + } + val reader = mock { + blocks.forEach { + if (it == 17964844L) { + on { + read(ChainRequest("chain_getBlockHash", ListParams(it.toHex()))) + } doReturn Mono.just(ChainResponse("\"$hash1\"".toByteArray(), null)) + on { + read(ChainRequest("state_getMetadata", ListParams(hash1))) + } doReturn Mono.just(ChainResponse(ByteArray(0), null)) + } else { + on { + read(ChainRequest("chain_getBlockHash", ListParams(it.toHex()))) + } doReturn Mono.just(ChainResponse("\"$hash2\"".toByteArray(), null)) + on { + read(ChainRequest("state_getMetadata", ListParams(hash2))) + } doReturn Mono.error(RuntimeException("State already discarded for")) + } + } + } + val upstream = mock { + on { getHead() } doReturn head + on { getIngressReader() } doReturn reader + } + + val detector = PolkadotLowerBoundService(Chain.UNSPECIFIED, upstream) StepVerifier.withVirtualTime { detector.detectLowerBounds() } .expectSubscription() @@ -49,7 +121,9 @@ class RecursiveLowerBoundServiceTest { assertThat(detector.getLowerBounds().toList()) .usingRecursiveFieldByFieldElementComparatorIgnoringFields("timestamp") .hasSameElementsAs( - listOf(LowerBoundData(17964844L, LowerBoundType.STATE)), + listOf( + LowerBoundData(17964844L, LowerBoundType.STATE), + ), ) } @@ -84,56 +158,6 @@ class RecursiveLowerBoundServiceTest { } companion object { - private val blocks = listOf( - 9000000L, 13500000L, 15750000L, 16875000L, 17437500L, 17718750L, 17859375L, 17929688L, 17964844L, 17947266L, - 17956055L, 17960449L, 17962646L, 17963745L, 17964294L, 17964569L, 17964706L, 17964775L, 17964809L, 17964826L, - 17964835L, 17964839L, 17964841L, 17964842L, 17964843L, - ) - private const val hash1 = "0x1b1a5dd69e12aa12e2b9197be0d0cceef3dde6368ea6376ad7c8b06488c9cf6a" - private const val hash2 = "0x1b1a5dd69e12aa12e2b9197be0d0cceef3dde6368ea6376ad7c8b06488c9cf7a" - - @JvmStatic - fun detectors(): List = listOf( - Arguments.of( - mock { - blocks.forEach { - if (it == 17964844L) { - on { - read(ChainRequest("eth_getBalance", ListParams(ZERO_ADDRESS, it.toHex()))) - } doReturn Mono.just(ChainResponse(ByteArray(0), null)) - } else { - on { - read(ChainRequest("eth_getBalance", ListParams(ZERO_ADDRESS, it.toHex()))) - } doReturn Mono.error(RuntimeException("missing trie node")) - } - } - }, - EthereumLowerBoundService::class.java, - ), - Arguments.of( - mock { - blocks.forEach { - if (it == 17964844L) { - on { - read(ChainRequest("chain_getBlockHash", ListParams(it.toHex()))) - } doReturn Mono.just(ChainResponse("\"$hash1\"".toByteArray(), null)) - on { - read(ChainRequest("state_getMetadata", ListParams(hash1))) - } doReturn Mono.just(ChainResponse(ByteArray(0), null)) - } else { - on { - read(ChainRequest("chain_getBlockHash", ListParams(it.toHex()))) - } doReturn Mono.just(ChainResponse("\"$hash2\"".toByteArray(), null)) - on { - read(ChainRequest("state_getMetadata", ListParams(hash2))) - } doReturn Mono.error(RuntimeException("State already discarded for")) - } - } - }, - PolkadotLowerBoundService::class.java, - ), - ) - @JvmStatic fun detectorsFirstBlock(): List = listOf( Arguments.of(