Beacon chain height matcher (#640)

* upgrade protoc (fixes local ./gradlew run issue)

* implement beacon chain block matcher

gradle/libs.versions.toml

@@ -1,7 +1,7 @@
 [versions]
 detekt = "1.23.1"
 groovy = "4.0.15"
-protoc = "3.21.7"
+protoc = "4.29.2"
 jackson = "2.11.0"
 grpc = "1.57.0"
 reactive-grpc = "1.2.0"

src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt

@@ -0,0 +1,64 @@
+package io.emeraldpay.dshackle.upstream.beaconchain
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.module.kotlin.readValue
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.upstream.ChainCallError
+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.rpcclient.RestParams
+import reactor.core.publisher.Flux
+import reactor.kotlin.core.publisher.toFlux
+
+class BeaconChainLowerBoundBlockDetector(
+    private val chain: Chain,
+    private val upstream: Upstream,
+) : LowerBoundDetector(chain) {
+    private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, stateErrors, lowerBounds)
+
+    companion object {
+        val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404}
+        val notFoundError2 = "Could not find requested block" // {"message":"Could not find requested block: signed beacon block can't be nil","code":404}
+        val stateErrors = setOf(notFoundError, notFoundError2)
+    }
+
+    override fun period(): Long {
+        return 5
+    }
+
+    override fun internalDetectLowerBound(): Flux<LowerBoundData> {
+        return recursiveLowerBound.recursiveDetectLowerBound { block ->
+            val restParams = RestParams(emptyList(), emptyList(), listOf(block.toString()), ByteArray(0))
+
+            upstream.getIngressReader()
+                .read(ChainRequest("GET#/eth/v2/beacon/blocks/*", restParams))
+                .flatMap(ChainResponse::requireResult)
+                .timeout(Defaults.internalCallsTimeout)
+                .map {
+                    parseHeadersResponse(it)
+                }
+        }.toFlux()
+    }
+
+    override fun types(): Set<LowerBoundType> {
+        return setOf(LowerBoundType.BLOCK)
+    }
+
+    private fun parseHeadersResponse(data: ByteArray): ChainResponse {
+        val node = Global.objectMapper.readValue<JsonNode>(data)
+        if (node.get("code") != null && node.get("message") != null && node.get("code").textValue() == "404") {
+            return ChainResponse(null, ChainCallError(node.get("code").asInt(), node.get("message").asText(), node.get("message").asText()))
+        }
+        if (node.get("data")?.get("message")?.get("slot") != null) {
+            return ChainResponse(node.get("data").toString().toByteArray(), null)
+        }
+        return ChainResponse(null, ChainCallError(404, notFoundError))
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt

@@ -7,9 +7,9 @@ import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
 
 class BeaconChainLowerBoundService(
     private val chain: Chain,
-    upstream: Upstream,
+    private val upstream: Upstream,
 ) : LowerBoundService(chain, upstream) {
     override fun detectors(): List<LowerBoundDetector> {
-        return listOf(BeaconChainLowerBoundStateDetector(chain))
+        return listOf(BeaconChainLowerBoundBlockDetector(chain, upstream))
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt

@@ -1,24 +0,0 @@
-package io.emeraldpay.dshackle.upstream.beaconchain
-
-import io.emeraldpay.dshackle.Chain
-import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
-import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector
-import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
-import reactor.core.publisher.Flux
-
-class BeaconChainLowerBoundStateDetector(
-    private val chain: Chain,
-) : LowerBoundDetector(chain) {
-
-    override fun period(): Long {
-        return 120
-    }
-
-    override fun internalDetectLowerBound(): Flux<LowerBoundData> {
-        return Flux.just(LowerBoundData(1, LowerBoundType.STATE))
-    }
-
-    override fun types(): Set<LowerBoundType> {
-        return setOf(LowerBoundType.STATE)
-    }
-}