Logs lower bound + fix state detection with null (#503)

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
         }
     }
 

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)
                         }
                     }
             }

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<LowerBoundData> {
+        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<LowerBoundType> {
+        return setOf(LowerBoundType.LOGS)
+    }
+}

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

@@ -14,6 +14,7 @@ class EthereumLowerBoundService(
             EthereumLowerBoundStateDetector(upstream),
             EthereumLowerBoundBlockDetector(upstream),
             EthereumLowerBoundTxDetector(upstream),
+            EthereumLowerBoundLogsDetector(upstream),
         )
     }
 }

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")
+                }
             }
         }
     }

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)
                     }
                 }
         }

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())

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
     }
 }

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<String, String>("client_type", "bor"),
+                        new Pair<String, String>("client_version", "v0.4.0"),
+                        new Pair<String, String>("archive", "false")
+                )
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
     def "Only default label"() {
         setup:
         def up = Mock(DefaultUpstream) {

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),
                 ),
             )
     }