Beacon lower bounds (state, blob, epoch) (#660)

* fix bound calculation when all responses were errors

* Add beacon chain blob,state,epoch bounds

src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt

@@ -148,6 +148,8 @@ class ChainEventMapper {
             LowerBoundType.LOGS -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS
             LowerBoundType.TRACE -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TRACE
             LowerBoundType.PROOF -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF
+            LowerBoundType.BLOB -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB
+            LowerBoundType.EPOCH -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH
         }
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt

@@ -108,6 +108,8 @@ class StreamHead(
             LowerBoundType.LOGS -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS
             LowerBoundType.TRACE -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TRACE
             LowerBoundType.PROOF -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF
+            LowerBoundType.BLOB -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB
+            LowerBoundType.EPOCH -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH
         }
     }
 

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

@@ -0,0 +1,66 @@
+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 BeaconChainLowerBoundBlobDetector(
+    private val chain: Chain,
+    private val upstream: Upstream,
+) : LowerBoundDetector(chain) {
+    private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOB, stateErrors, lowerBounds)
+
+    companion object {
+        const val MAX_OFFSET = 20
+        val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404}
+        val notFoundError2 = "lock not found"
+        val notFoundError3 = "has not been found"
+        val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3)
+    }
+
+    override fun period(): Long {
+        return 5
+    }
+
+    override fun internalDetectLowerBound(): Flux<LowerBoundData> {
+        return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { block ->
+            val restParams = RestParams(emptyList(), emptyList(), listOf(block.toString()), ByteArray(0))
+
+            upstream.getIngressReader()
+                .read(ChainRequest("GET#/eth/v1/beacon/blob_sidecars/*", restParams))
+                .flatMap(ChainResponse::requireResult)
+                .timeout(Defaults.internalCallsTimeout)
+                .map {
+                    parseHeadersResponse(it)
+                }
+        }.toFlux()
+    }
+
+    override fun types(): Set<LowerBoundType> {
+        return setOf(LowerBoundType.BLOB)
+    }
+
+    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").toString() == "[]") {
+            return ChainResponse(null, ChainCallError(404, notFoundError))
+        }
+        return ChainResponse(node.get("data").toString().toByteArray(), null)
+    }
+}

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

@@ -24,11 +24,13 @@ class BeaconChainLowerBoundBlockDetector(
     private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.BLOCK, stateErrors, lowerBounds)
 
     companion object {
-        const val MAX_OFFSET = 30
+        const val MAX_OFFSET = 20
         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 notFoundError3 = "has not been found" // Block header/data has not been found
-        val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3)
+        val notFoundError4 = "lock not found" // {"message":"block not found 1413","code":404}
+        val notFoundError5 = "Internal Server Error" // block pi returns {"message":"Internal Server Error"} in first 9 blocks (?)
+        val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3, notFoundError4, notFoundError5)
     }
 
     override fun period(): Long {

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

@@ -0,0 +1,96 @@
+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.LowerBounds
+import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound
+import io.emeraldpay.dshackle.upstream.rpcclient.RestParams
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.kotlin.core.publisher.toFlux
+
+class EpochRecursiveLowerBound(
+    upstream: Upstream,
+    type: LowerBoundType,
+    nonRetryableErrors: Set<String>,
+    lowerBounds: LowerBounds,
+) : RecursiveLowerBound(upstream, type, nonRetryableErrors, lowerBounds) {
+
+    // similar to recursive lower bound, but range is adjusted as epoch range is 32x smaller than height
+    override fun initialRange(): Mono<LowerBoundBinarySearchData> {
+        return Mono.just(upstream.getHead())
+            .flatMap {
+                val currentHeight = it.getCurrentHeight()
+                if (currentHeight == null) {
+                    Mono.empty()
+                } else if (lowerBounds.getLastBound(type) == null) {
+                    Mono.just(LowerBoundBinarySearchData(0, currentHeight / 32)) // 1 epoch is 32 slots
+                } else {
+                    Mono.just(LowerBoundBinarySearchData(lowerBounds.getLastBound(type)!!.lowerBound, currentHeight / 32)) // 1 epoch is 32 slots
+                }
+            }
+    }
+}
+
+class BeaconChainLowerBoundEpochDetector(
+    private val chain: Chain,
+    private val upstream: Upstream,
+) : LowerBoundDetector(chain) {
+    private val recursiveLowerBound = EpochRecursiveLowerBound(upstream, LowerBoundType.EPOCH, stateErrors, lowerBounds)
+
+    companion object {
+        const val MAX_OFFSET = 20
+        val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404}
+        val notFoundError2 = "Could not get requested state"
+        val notFoundError3 = "missing state" // "missing state at slot 11609023"
+        val stateErrors = setOf(notFoundError, notFoundError2, notFoundError3)
+    }
+
+    override fun period(): Long {
+        return 5
+    }
+
+    override fun internalDetectLowerBound(): Flux<LowerBoundData> {
+        return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { slot ->
+            val restParams = RestParams(listOf(), emptyList(), listOf(slot.toString()), "[\"1\"]".toByteArray())
+
+            upstream.getIngressReader()
+                .read(ChainRequest("POST#/eth/v1/beacon/rewards/attestations/*", restParams))
+                .flatMap(ChainResponse::requireResult)
+                .timeout(Defaults.internalCallsTimeout)
+                .map {
+                    parseHeadersResponse(it)
+                }
+        }.toFlux()
+    }
+
+    override fun types(): Set<LowerBoundType> {
+        return setOf(LowerBoundType.EPOCH)
+    }
+
+    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()))
+        }
+
+        val jsonData = node.get("data")
+        if (jsonData != null) {
+            val str = jsonData.toString()
+            if (str.length >= 2) {
+                return ChainResponse(str.toByteArray(), null)
+            }
+        }
+        return ChainResponse(null, ChainCallError(404, notFoundError))
+    }
+}

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

@@ -10,6 +10,11 @@ class BeaconChainLowerBoundService(
     private val upstream: Upstream,
 ) : LowerBoundService(chain, upstream) {
     override fun detectors(): List<LowerBoundDetector> {
-        return listOf(BeaconChainLowerBoundBlockDetector(chain, upstream))
+        return listOf(
+            BeaconChainLowerBoundBlockDetector(chain, upstream),
+            BeaconChainLowerBoundEpochDetector(chain, upstream),
+            BeaconChainLowerBoundStateDetector(chain, upstream),
+            BeaconChainLowerBoundBlobDetector(chain, upstream),
+        )
     }
 }

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

@@ -0,0 +1,70 @@
+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 BeaconChainLowerBoundStateDetector(
+    private val chain: Chain,
+    private val upstream: Upstream,
+) : LowerBoundDetector(chain) {
+    private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, stateErrors, lowerBounds)
+
+    companion object {
+        const val MAX_OFFSET = 20
+        val notFoundError = "NOT_FOUND:" // e.g. {"message":"NOT_FOUND: beacon block at slot 1086646","code":404}
+        val notFoundError2 = "Could not get requested state"
+        val stateErrors = setOf(notFoundError, notFoundError2)
+    }
+
+    override fun period(): Long {
+        return 5
+    }
+
+    override fun internalDetectLowerBound(): Flux<LowerBoundData> {
+        return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { slot ->
+            val restParams = RestParams(listOf(), emptyList(), listOf(slot.toString()), "[\"1\"]".toByteArray())
+
+            upstream.getIngressReader()
+                .read(ChainRequest("POST#/eth/v1/beacon/states/*/validator_balances", restParams))
+                .flatMap(ChainResponse::requireResult)
+                .timeout(Defaults.internalCallsTimeout)
+                .map {
+                    parseHeadersResponse(it)
+                }
+        }.toFlux()
+    }
+
+    override fun types(): Set<LowerBoundType> {
+        return setOf(LowerBoundType.STATE)
+    }
+
+    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()))
+        }
+
+        val jsonData = node.get("data")
+        if (jsonData != null) {
+            val str = jsonData.toString()
+            if (str.length >= 2) {
+                return ChainResponse(str.toByteArray(), null)
+            }
+        }
+        return ChainResponse(null, ChainCallError(404, notFoundError))
+    }
+}

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, LOGS, TRACE, PROOF
+    UNKNOWN, STATE, SLOT, BLOCK, TX, LOGS, TRACE, PROOF, BLOB, EPOCH
 }
 
 fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType {
@@ -34,7 +34,7 @@ fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType {
         BlockchainOuterClass.LowerBoundType.LOWER_BOUND_LOGS -> LowerBoundType.LOGS
         BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TRACE -> LowerBoundType.TRACE
         BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF -> LowerBoundType.PROOF
-        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB -> LowerBoundType.UNKNOWN
+        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB -> LowerBoundType.BLOB
-        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH -> LowerBoundType.UNKNOWN
+        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH -> LowerBoundType.EPOCH
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt

@@ -14,13 +14,13 @@ import reactor.util.retry.RetryBackoffSpec
 import java.time.Duration
 import java.util.concurrent.atomic.AtomicInteger
 
-class RecursiveLowerBound(
+open class RecursiveLowerBound(
-    private val upstream: Upstream,
+    protected val upstream: Upstream,
-    private val type: LowerBoundType,
+    protected val type: LowerBoundType,
-    private val nonRetryableErrors: Set<String>,
+    protected val nonRetryableErrors: Set<String>,
-    private val lowerBounds: LowerBounds,
+    protected val lowerBounds: LowerBounds,
 ) {
-    private val log = LoggerFactory.getLogger(this::class.java)
+    protected val log = LoggerFactory.getLogger(this::class.java)
 
     fun recursiveDetectLowerBound(hasData: (Long) -> Mono<ChainResponse>): Flux<LowerBoundData> {
         return initialRange()
@@ -32,7 +32,17 @@ class RecursiveLowerBound(
 
                     if (data.left > data.right) {
                         val current = if (data.current == 0L) 1 else data.current
-                        Mono.just(LowerBoundBinarySearchData(current, true))
+                        hasData(current)
+                            .retryWhen(retrySpec(middle, nonRetryableErrors))
+                            .flatMap(ChainResponse::requireResult)
+                            .map { LowerBoundBinarySearchData(current, true) }
+                            .onErrorResume {
+                                if (current == 1L && data.right > 10) {
+                                    Mono.empty() // Couldn't detect bound: data.right is chain height here and current wasn't set because we got errors all the time
+                                } else {
+                                    Mono.just(LowerBoundBinarySearchData(current, true)) // if we approached left bound(1) or node have just pruned data.current due to long bound calculation, return data.current as is
+                                }
+                            }
                     } else {
                         hasData(middle)
                             .retryWhen(retrySpec(middle, nonRetryableErrors))
@@ -75,7 +85,17 @@ class RecursiveLowerBound(
 
                             if (data.left > data.right) {
                                 val current = if (data.current == 0L) 1 else data.current
-                                Mono.just(LowerBoundBinarySearchData(current, true))
+                                hasData(current)
+                                    .retryWhen(retrySpec(middle, nonRetryableErrors))
+                                    .flatMap(ChainResponse::requireResult)
+                                    .map { LowerBoundBinarySearchData(current, true) }
+                                    .onErrorResume {
+                                        if (current == 1L && data.right > 10) {
+                                            Mono.empty() // Couldn't detect bound: data.right is chain height here and current wasn't set because we got errors all the time
+                                        } else {
+                                            Mono.just(LowerBoundBinarySearchData(current, true)) // if we approached left bound(1) or node have just pruned data.current due to long bound calculation, return data.current as is
+                                        }
+                                    }
                             } else {
                                 hasData(middle)
                                     .retryWhen(retrySpec(middle, nonRetryableErrors))
@@ -99,7 +119,7 @@ class RecursiveLowerBound(
             )
     }
 
-    private fun shiftLeftAndSearch(
+    protected fun shiftLeftAndSearch(
         currentData: LowerBoundBinarySearchData,
         currentMiddle: Long,
         visitedBlocks: HashSet<Long>,
@@ -145,7 +165,7 @@ class RecursiveLowerBound(
             }
     }
 
-    private fun initialRange(): Mono<LowerBoundBinarySearchData> {
+    protected open fun initialRange(): Mono<LowerBoundBinarySearchData> {
         return Mono.just(upstream.getHead())
             .flatMap {
                 val currentHeight = it.getCurrentHeight()
@@ -160,7 +180,7 @@ class RecursiveLowerBound(
             }
     }
 
-    private fun retrySpec(block: Long, nonRetryableErrors: Set<String>): RetryBackoffSpec {
+    protected fun retrySpec(block: Long, nonRetryableErrors: Set<String>): RetryBackoffSpec {
         return Retry.backoff(
             Long.MAX_VALUE,
             Duration.ofSeconds(1),
@@ -172,8 +192,9 @@ class RecursiveLowerBound(
             .doAfterRetry {
                 if (it.totalRetries() > 30) {
                     log.warn(
-                        "There are too much retries to calculate {} lower bound of upstream {}, " +
+                        "There are too much retries to calculate {} lower bound of upstream {}, block {} " +
                             "probably this error with message `{}` is not retryable, please report it to dshackle devs",
+                        block,
                         type,
                         upstream.getId(),
                         it.failure().message,
@@ -191,10 +212,10 @@ class RecursiveLowerBound(
             }
     }
 
-    private fun middleBlock(lowerBoundBinarySearchData: LowerBoundBinarySearchData): Long =
+    protected fun middleBlock(lowerBoundBinarySearchData: LowerBoundBinarySearchData): Long =
         lowerBoundBinarySearchData.left + (lowerBoundBinarySearchData.right - lowerBoundBinarySearchData.left) / 2
 
-    private data class LowerBoundBinarySearchData(
+    protected data class LowerBoundBinarySearchData(
         val left: Long,
         val right: Long,
         val current: Long,