From f9667e2f60b75e72c5c149889b5c7458a0997214 Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Tue, 11 Jun 2024 14:13:18 +0400 Subject: [PATCH] Logs lower bound + fix state detection with null (#503) --- emerald-grpc | 2 +- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 1 + .../EthereumLowerBoundBlockDetector.kt | 9 +++- .../EthereumLowerBoundLogsDetector.kt | 53 +++++++++++++++++++ .../ethereum/EthereumLowerBoundService.kt | 1 + .../EthereumLowerBoundStateDetector.kt | 6 +++ .../ethereum/EthereumLowerBoundTxDetector.kt | 12 +++-- .../EthereumUpstreamSettingsDetector.kt | 9 +++- .../upstream/lowerbound/LowerBoundData.kt | 3 +- ...thereumUpstreamSettingsDetectorSpec.groovy | 28 +++++++++- .../RecursiveLowerBoundServiceTest.kt | 10 +++- 11 files changed, 122 insertions(+), 12 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundLogsDetector.kt diff --git a/emerald-grpc b/emerald-grpc index 31527635..a3351cdc 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 315276353fefc556e47fa3d0f1a191f7522f784e +Subproject commit a3351cdcb5448f422dc61726a5186c6a16c8829a diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index df9b14e3..5f4f92a7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -97,6 +97,7 @@ class StreamHead( LowerBoundType.STATE -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_STATE LowerBoundType.BLOCK -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOCK LowerBoundType.TX -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TX + LowerBoundType.LOGS -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt index b00051ff..257dd98b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt @@ -15,7 +15,12 @@ import reactor.core.publisher.Mono class EthereumLowerBoundBlockDetector( private val upstream: Upstream, ) : LowerBoundDetector() { - private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, setOf("No block data"), lowerBounds) + + companion object { + private const val NO_BLOCK_DATA = "No block data" + } + + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, setOf(NO_BLOCK_DATA), lowerBounds) override fun period(): Long { return 3 @@ -35,7 +40,7 @@ class EthereumLowerBoundBlockDetector( ) .doOnNext { if (it.hasResult() && it.getResult().contentEquals("null".toByteArray())) { - throw IllegalStateException("No block data") + throw IllegalStateException(NO_BLOCK_DATA) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundLogsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundLogsDetector.kt new file mode 100644 index 00000000..4b2a1308 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundLogsDetector.kt @@ -0,0 +1,53 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.upstream.ChainRequest +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 + +class EthereumLowerBoundLogsDetector( + private val upstream: Upstream, +) : LowerBoundDetector() { + + companion object { + const val MAX_OFFSET = 20 + private const val NO_LOGS_DATA = "No logs data" + } + + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.LOGS, setOf(NO_LOGS_DATA), lowerBounds) + + override fun period(): Long { + return 3 + } + + override fun internalDetectLowerBound(): Flux { + return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { block -> + upstream.getIngressReader() + .read( + ChainRequest( + "eth_getLogs", + ListParams( + mapOf( + "fromBlock" to block.toHex(), + "toBlock" to block.toHex(), + ), + ), + ), + ) + .doOnNext { + if (it.hasResult() && (it.getResult().contentEquals("null".toByteArray()) || it.getResult().contentEquals("[]".toByteArray()))) { + throw IllegalStateException(NO_LOGS_DATA) + } + } + } + } + + override fun types(): Set { + return setOf(LowerBoundType.LOGS) + } +} 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 80b122c2..b6671387 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt @@ -14,6 +14,7 @@ class EthereumLowerBoundService( EthereumLowerBoundStateDetector(upstream), EthereumLowerBoundBlockDetector(upstream), EthereumLowerBoundTxDetector(upstream), + EthereumLowerBoundLogsDetector(upstream), ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt index 4c0282fe..9587b9fd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.Upstream @@ -41,6 +42,7 @@ class EthereumLowerBoundStateDetector( "historical backend error", // optimism "load state tree: failed to load state tree", // filecoin "purged for block", // erigon + "No state data", // our own error if there is "null" in response ) } @@ -59,6 +61,10 @@ class EthereumLowerBoundStateDetector( ListParams(ZERO_ADDRESS, block.toHex()), ), ) + }.doOnNext { + if (it.hasResult() && it.getResult().contentEquals(Global.nullValue)) { + throw IllegalStateException("No state data") + } } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt index 36c7d8b6..5e855c41 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt @@ -10,12 +10,16 @@ import io.emeraldpay.dshackle.upstream.lowerbound.toHex import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import reactor.core.publisher.Flux -const val MAX_OFFSET = 20 - class EthereumLowerBoundTxDetector( private val upstream: Upstream, ) : LowerBoundDetector() { - private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.TX, setOf("No tx data"), lowerBounds) + + companion object { + const val MAX_OFFSET = 20 + private const val NO_TX_DATA = "No tx data" + } + + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.TX, setOf(NO_TX_DATA), lowerBounds) override fun period(): Long { return 3 @@ -32,7 +36,7 @@ class EthereumLowerBoundTxDetector( ) .doOnNext { if (it.hasResult() && (it.getResult().contentEquals("null".toByteArray()) || it.getResultAsProcessedString().substring(2).toLong(16) == 0L)) { - throw IllegalStateException("No tx data") + throw IllegalStateException(NO_TX_DATA) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt index 24c93405..6225c5fc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import com.fasterxml.jackson.databind.JsonNode import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.BasicEthUpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse @@ -60,7 +61,13 @@ class EthereumUpstreamSettingsDetector( "eth_getBalance", ListParams(ZERO_ADDRESS, blockNumber), ), - ).flatMap(ChainResponse::requireResult) + ) + .flatMap(ChainResponse::requireResult) + .doOnNext { + if (it.contentEquals(Global.nullValue)) { + throw IllegalStateException("Null data") + } + } } override fun nodeTypeRequest(): NodeTypeRequest = NodeTypeRequest(clientVersionRequest()) 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 99e3e473..c297469a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt @@ -20,7 +20,7 @@ data class LowerBoundData( } enum class LowerBoundType { - UNKNOWN, STATE, SLOT, BLOCK, TX + UNKNOWN, STATE, SLOT, BLOCK, TX, LOGS } fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType { @@ -31,5 +31,6 @@ fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType { BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOCK -> LowerBoundType.BLOCK BlockchainOuterClass.LowerBoundType.UNRECOGNIZED -> LowerBoundType.UNKNOWN BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TX -> LowerBoundType.TX + BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS -> LowerBoundType.LOGS } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetectorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetectorSpec.groovy index 29747187..7353aed1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetectorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetectorSpec.groovy @@ -1,13 +1,12 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.ApiReaderMock import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import kotlin.Pair import reactor.core.publisher.Mono @@ -49,6 +48,31 @@ class EthereumUpstreamSettingsDetectorSpec extends Specification { "Bor/v0.4.0/linux-amd64/go1.19.10" | "bor" | "v0.4.0" } + def "Not archival node if null response"() { + setup: + def up = TestingCommons.upstream( + new ApiReaderMock().tap { + answer("web3_clientVersion", [], "Bor/v0.4.0/linux-amd64/go1.19.10") + answer("eth_blockNumber", [], "0x10df3e5") + answer("eth_getBalance", ["0x0000000000000000000000000000000000000000", "0x10dccd5"], "") + answer("eth_getBalance", ["0x0000000000000000000000000000000000000000", "0x2710"], null) + } + ) + def detector = new EthereumUpstreamSettingsDetector(up, Chain.ETHEREUM__MAINNET) + + when: + def act = detector.detectLabels() + then: + StepVerifier.create(act) + .expectNext( + new Pair("client_type", "bor"), + new Pair("client_version", "v0.4.0"), + new Pair("archive", "false") + ) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + def "Only default label"() { setup: def up = Mock(DefaultUpstream) { diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt index d9d9541c..ff39c5db 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt @@ -3,7 +3,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundService -import io.emeraldpay.dshackle.upstream.ethereum.MAX_OFFSET +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundTxDetector.Companion.MAX_OFFSET import io.emeraldpay.dshackle.upstream.ethereum.ZERO_ADDRESS import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService @@ -49,6 +49,9 @@ class RecursiveLowerBoundServiceTest { on { read(ChainRequest("eth_getBlockTransactionCountByNumber", ListParams(it.toHex()))) } doReturn Mono.just(ChainResponse("\"0x12\"".toByteArray(), null)) + on { + read(ChainRequest("eth_getLogs", ListParams(mapOf("fromBlock" to it.toHex(), "toBlock" to it.toHex())))) + } doReturn Mono.just(ChainResponse("[\"0x12\"]".toByteArray(), null)) } else { on { read(ChainRequest("eth_getBalance", ListParams(ZERO_ADDRESS, it.toHex()))) @@ -60,6 +63,9 @@ class RecursiveLowerBoundServiceTest { on { read(ChainRequest("eth_getBlockTransactionCountByNumber", ListParams(block.toHex()))) } doReturn Mono.error(RuntimeException("No tx data")) + on { + read(ChainRequest("eth_getLogs", ListParams(mapOf("fromBlock" to block.toHex(), "toBlock" to block.toHex())))) + } doReturn Mono.error(RuntimeException("No logs data")) } } } @@ -77,6 +83,7 @@ class RecursiveLowerBoundServiceTest { .expectNextMatches { it.lowerBound == 17964844L && it.type == LowerBoundType.STATE } .expectNextMatches { it.lowerBound == 17964844L && it.type == LowerBoundType.BLOCK } .expectNextMatches { it.lowerBound == 17964844L && it.type == LowerBoundType.TX } + .expectNextMatches { it.lowerBound == 17964844L && it.type == LowerBoundType.LOGS } .thenCancel() .verify(Duration.ofSeconds(3)) @@ -87,6 +94,7 @@ class RecursiveLowerBoundServiceTest { LowerBoundData(17964844L, LowerBoundType.STATE), LowerBoundData(17964844L, LowerBoundType.BLOCK), LowerBoundData(17964844L, LowerBoundType.TX), + LowerBoundData(17964844L, LowerBoundType.LOGS), ), ) }