Reanimate grpc upstreams (#491)

2024-05-30 14:03:53 +04:00
parent f3ab6b3443
commit 71d68e858b
18 changed files with 237 additions and 60 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt
@@ -10,6 +10,7 @@ import org.springframework.context.annotation.Configuration
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory
 import reactor.core.scheduler.Scheduler
 import reactor.core.scheduler.Schedulers
+import java.util.concurrent.Executor
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors

@@ -63,6 +64,11 @@ open class SchedulersConfig {
        return makeScheduler("head-liveness-scheduler", 4, monitoringConfig)
    }

+    @Bean
+    open fun grpcChannelExecutor(monitoringConfig: MonitoringConfig): Executor {
+        return makePool("grpc-client-channel", 10, monitoringConfig)
+    }
+
    @Bean
    open fun authScheduler(monitoringConfig: MonitoringConfig): Scheduler {
        return makeScheduler("auth-scheduler", 4, monitoringConfig)
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.startup

 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.config.ChainsConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.foundation.ChainOptions
 import io.emeraldpay.dshackle.startup.configure.UpstreamCreationData
@@ -33,6 +34,7 @@ open class ConfiguredUpstreams(
    private val upstreamFactory: UpstreamFactory,
    private val config: UpstreamsConfig,
    private val multistreamHolder: CurrentMultistreamHolder,
+    private val chainsConfig: ChainsConfig,
 ) : ApplicationRunner {
    private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)

@@ -71,6 +73,19 @@ open class ConfiguredUpstreams(
                            ),
                        )
                }
+            } else {
+                upstreamFactory.createGrpcUpstream(
+                    up as UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
+                    chainsConfig,
+                )
+                    .start()
+                    .doOnNext {
+                        log.info("Chain ${it.chain} ${it.type} through gRPC at ${up.connection?.host}:${up.connection?.port}. With caps: ${it.upstream.getCapabilities()}")
+                    }
+                    .subscribe {
+                        multistreamHolder.getUpstream(it.chain)
+                            .processUpstreamsEvents(it)
+                    }
            }
        }
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt
@@ -13,8 +13,6 @@ import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry
 import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
 import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory
 import org.springframework.stereotype.Component
-import java.util.function.Function
-import kotlin.math.abs

@Component
 open class GenericUpstreamCreator(
@@ -72,7 +70,7 @@ open class GenericUpstreamCreator(
        val hashUrl = connection.let {
            if (it.connectorMode == GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
        }
-        val hash = getHash(nodeId, hashUrl!!)
+        val hash = getHash(nodeId, hashUrl!!, hashes)

        val upstream = GenericUpstream(
            config.id!!,
@@ -96,25 +94,4 @@ open class GenericUpstreamCreator(
        }
        return UpstreamCreationData(upstream, upstream.isValid())
    }
-
-    private fun getHash(nodeId: Int?, obj: Any): Byte =
-        nodeId?.toByte() ?: (obj.hashCode() % 255).let {
-            if (it == 0) 1 else it
-        }.let { nonZeroHash ->
-            listOf<Function<Int, Int>>(
-                Function { i -> i },
-                Function { i -> (-i) },
-                Function { i -> 127 - abs(i) },
-                Function { i -> abs(i) - 128 },
-            ).map {
-                it.apply(nonZeroHash).toByte()
-            }.firstOrNull {
-                hashes[it] != true
-            }?.let {
-                hashes[it] = true
-                it
-            } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
-                it != 0 && hashes[it.toByte()] != true
-            }.toByte()
-        }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt
@@ -11,6 +11,8 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
+import java.util.function.Function
+import kotlin.math.abs

 abstract class UpstreamCreator(
    private val chainsConfig: ChainsConfig,
@@ -19,6 +21,29 @@ abstract class UpstreamCreator(
 ) {
    protected val log: Logger = LoggerFactory.getLogger(this::class.java)

+    companion object {
+        fun getHash(nodeId: Int?, obj: Any, hashes: MutableMap<Byte, Boolean>): Byte =
+            nodeId?.toByte() ?: (obj.hashCode() % 255).let {
+                if (it == 0) 1 else it
+            }.let { nonZeroHash ->
+                listOf<Function<Int, Int>>(
+                    Function { i -> i },
+                    Function { i -> (-i) },
+                    Function { i -> 127 - abs(i) },
+                    Function { i -> abs(i) - 128 },
+                ).map {
+                    it.apply(nonZeroHash).toByte()
+                }.firstOrNull {
+                    hashes[it] != true
+                }?.let {
+                    hashes[it] = true
+                    it
+                } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
+                    it != 0 && hashes[it.toByte()] != true
+                }.toByte()
+            }
+    }
+
    fun createUpstream(
        upstreamsConfig: UpstreamsConfig.Upstream<*>,
        defaultOptions: Map<Chain, ChainOptions.PartialOptions>,
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt
@@ -2,9 +2,12 @@ package io.emeraldpay.dshackle.startup.configure

 import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.config.ChainsConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.foundation.ChainOptions
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreamCreator
+import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
 import org.springframework.stereotype.Component

 data class UpstreamCreationData(
@@ -21,6 +24,7 @@ class UpstreamFactory(
    private val genericUpstreamCreator: GenericUpstreamCreator,
    private val ethereumUpstreamCreator: EthereumUpstreamCreator,
    private val bitcoinUpstreamCreator: BitcoinUpstreamCreator,
+    private val grpcUpstreamCreator: GrpcUpstreamCreator,
 ) {

    fun createUpstream(
@@ -34,4 +38,11 @@ class UpstreamFactory(
            else -> genericUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions)
        }
    }
+
+    fun createGrpcUpstream(
+        config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
+        chainsConfig: ChainsConfig,
+    ): GrpcUpstreams {
+        return grpcUpstreamCreator.creatGrpcUpstream(config, chainsConfig)
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.upstream

 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.config.ChainsConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.foundation.ChainOptions
@@ -39,6 +40,7 @@ abstract class DefaultUpstream(
    private val targets: CallMethods?,
    private val node: QuorumForLabels.QuorumItem?,
    private val chainConfig: ChainsConfig.ChainConfig,
+    private val chain: Chain,
 ) : Upstream {

    constructor(
@@ -49,8 +51,9 @@ abstract class DefaultUpstream(
        targets: CallMethods?,
        node: QuorumForLabels.QuorumItem?,
        chainConfig: ChainsConfig.ChainConfig,
+        chain: Chain,
    ) :
-        this(id, hash, null, UpstreamAvailability.UNAVAILABLE, options, role, targets, node, chainConfig)
+        this(id, hash, null, UpstreamAvailability.UNAVAILABLE, options, role, targets, node, chainConfig, chain)

    protected val log = LoggerFactory.getLogger(this::class.java)

@@ -159,5 +162,11 @@ abstract class DefaultUpstream(
        // NOOP
    }

+    protected fun sendUpstreamStateEvent(eventType: UpstreamChangeEvent.ChangeType) {
+        stateEventStream.emitNext(
+            UpstreamChangeEvent(chain, this, eventType),
+        ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+    }
+
    data class Status(val lag: Long?, val avail: UpstreamAvailability, val status: UpstreamAvailability)
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
@@ -27,7 +27,7 @@ import io.emeraldpay.dshackle.upstream.MatchesResponse.NotMatchedResponse
 import io.emeraldpay.dshackle.upstream.MatchesResponse.SameNodeResponse
 import io.emeraldpay.dshackle.upstream.MatchesResponse.SlotHeightResponse
 import io.emeraldpay.dshackle.upstream.MatchesResponse.Success
-import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
+import io.emeraldpay.dshackle.upstream.lowerbound.fromProtoType
 import org.apache.commons.lang3.StringUtils
 import java.util.Collections

@@ -76,7 +76,7 @@ class Selector {
                } else if (selector.hasLowerHeightSelector()) {
                    return Sort(
                        compareBy(nullsLast()) {
-                            it.getLowerBound(fromProtoType(selector.lowerHeightSelector.lowerBoundType))?.lowerBound
+                            it.getLowerBound(selector.lowerHeightSelector.lowerBoundType.fromProtoType())?.lowerBound
                        },
                    )
                }
@@ -84,16 +84,6 @@ class Selector {
            return Sort.default
        }

-        private fun fromProtoType(type: BlockchainOuterClass.LowerBoundType): LowerBoundType {
-            return when (type) {
-                BlockchainOuterClass.LowerBoundType.LOWER_BOUND_SLOT -> LowerBoundType.SLOT
-                BlockchainOuterClass.LowerBoundType.LOWER_BOUND_UNSPECIFIED -> LowerBoundType.UNKNOWN
-                BlockchainOuterClass.LowerBoundType.LOWER_BOUND_STATE -> LowerBoundType.STATE
-                BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOCK -> LowerBoundType.BLOCK
-                BlockchainOuterClass.LowerBoundType.UNRECOGNIZED -> LowerBoundType.UNKNOWN
-            }
-        }
-
        @JvmStatic
        fun convertToMatcher(req: BlockchainOuterClass.Selector?): LabelSelectorMatcher {
            return when {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
@@ -33,7 +33,7 @@ abstract class BitcoinUpstream(
    node: QuorumForLabels.QuorumItem,
    val esploraClient: EsploraClient? = null,
    chainConfig: ChainsConfig.ChainConfig,
-) : DefaultUpstream(id, 0.toByte(), options, role, callMethods, node, chainConfig) {
+) : DefaultUpstream(id, 0.toByte(), options, role, callMethods, node, chainConfig, chain) {

    constructor(
        id: String,
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt
@@ -68,7 +68,7 @@ open class GenericMultistream(
    }

    override fun addUpstreamInternal(u: Upstream) {
-        upstreams.add(u as GenericUpstream)
+        upstreams.add(u)
    }

    private val head: DynamicMergedHead = DynamicMergedHead(
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
@@ -29,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
-import reactor.core.publisher.Sinks
 import java.time.Duration
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
@@ -47,7 +46,7 @@ open class GenericUpstream(
    validatorBuilder: UpstreamValidatorBuilder,
    upstreamSettingsDetectorBuilder: UpstreamSettingsDetectorBuilder,
    lowerBoundServiceBuilder: LowerBoundServiceBuilder,
-) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig), Lifecycle {
+) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig, chain), Lifecycle {

    private val validator: UpstreamValidator? = validatorBuilder(chain, this, getOptions(), chainConfig)
    private var validatorSubscription: Disposable? = null
@@ -255,10 +254,4 @@ open class GenericUpstream(
    }

    fun isValid(): Boolean = isUpstreamValid.get()
-
-    private fun sendUpstreamStateEvent(eventType: UpstreamChangeEvent.ChangeType) {
-        stateEventStream.emitNext(
-            UpstreamChangeEvent(chain, this, eventType),
-        ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
-    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
@@ -72,7 +72,7 @@ class BitcoinGrpcUpstream(

    private val extractBlock = ExtractBlock()
    private val defaultReader: ChainReader = client.getReader()
-    private val blockConverter: Function<BlockchainOuterClass.ChainHead, BlockContainer> = Function { value ->
+    private val blockConverter: Function<BlockchainOuterClass.ChainHead, GrpcHead.GrpcHeadData> = Function { value ->
        val parentHash =
            if (value.parentBlockId.isBlank()) {
                null
@@ -89,7 +89,7 @@ class BitcoinGrpcUpstream(
            null,
            parentHash,
        )
-        block
+        GrpcHead.GrpcHeadData(block)
    }

    private val reloadBlock: Function<BlockContainer, Publisher<BlockContainer>> = Function { existingBlock ->
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt
@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.foundation.ChainOptions
 import io.emeraldpay.dshackle.reader.ChainReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.BuildInfo
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
@@ -37,12 +38,17 @@ import io.emeraldpay.dshackle.upstream.ethereum.domain.BlockHash
 import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
 import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
 import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
+import io.emeraldpay.dshackle.upstream.lowerbound.fromProtoType
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory
 import reactor.core.publisher.Flux
 import reactor.core.scheduler.Scheduler
+import reactor.core.scheduler.Schedulers
 import java.math.BigInteger
 import java.time.Instant
 import java.util.Locale
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.Executors
 import java.util.function.Function

 open class GenericGrpcUpstream(
@@ -64,11 +70,12 @@ open class GenericGrpcUpstream(
    null,
    null,
    chainConfig,
+    chain,
 ),
    GrpcUpstream,
    Lifecycle {

-    private val blockConverter: Function<BlockchainOuterClass.ChainHead, BlockContainer> = Function { value ->
+    private val blockConverter: Function<BlockchainOuterClass.ChainHead, GrpcHead.GrpcHeadData> = Function { value ->
        val parentHash =
            if (value.parentBlockId.isBlank()) {
                null
@@ -85,7 +92,9 @@ open class GenericGrpcUpstream(
            null,
            parentHash,
        )
-        block
+        val lowerBounds = value.lowerBoundsList
+            .map { LowerBoundData(it.lowerBoundValue, it.lowerBoundTimestamp, it.lowerBoundType.fromProtoType()) }
+        GrpcHead.GrpcHeadData(block, lowerBounds)
    }

    private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
@@ -104,7 +113,15 @@ open class GenericGrpcUpstream(

    private val defaultReader: ChainReader = client.getReader()

+    private val lowerBounds = ConcurrentHashMap<LowerBoundType, LowerBoundData>()
+
    override fun start() {
+        grpcHead.lowerBoundsFlux()
+            .publishOn(lowerBoundScheduler)
+            .subscribe {
+                lowerBounds[it.type] = it
+                sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.UPDATED)
+            }
    }

    override fun isRunning(): Boolean {
@@ -181,14 +198,24 @@ open class GenericGrpcUpstream(
    }

    override fun getLowerBounds(): Collection<LowerBoundData> {
-        return emptyList()
+        return lowerBounds.values
    }

    override fun getLowerBound(lowerBoundType: LowerBoundType): LowerBoundData? {
-        return null
+        return lowerBounds[lowerBoundType]
    }

    override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {
-        return null
+        return Upstream.UpstreamSettingsData(
+            nodeId(),
+            getId(),
+            "unknown",
+        )
+    }
+
+    companion object {
+        val lowerBoundScheduler: Scheduler = Schedulers.fromExecutorService(
+            Executors.newFixedThreadPool(4, CustomizableThreadFactory("grpc-lower-bound-")),
+        )
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
@@ -25,12 +25,14 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
 import io.micrometer.core.instrument.Counter
 import io.micrometer.core.instrument.Metrics
 import org.reactivestreams.Publisher
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.publisher.Sinks
 import reactor.core.scheduler.Scheduler
 import reactor.kotlin.extra.retry.retryExponentialBackoff
 import java.time.Duration
@@ -44,7 +46,7 @@ class GrpcHead(
    /**
     * Converted from remote head details to the block container, which could be partial at this point
     */
-    private val converter: Function<BlockchainOuterClass.ChainHead, BlockContainer>,
+    private val converter: Function<BlockchainOuterClass.ChainHead, GrpcHeadData>,
    /**
     * Populate block data with all missing details, of any
     */
@@ -55,6 +57,8 @@ class GrpcHead(

    private var headSubscription: Disposable? = null

+    private val lowerBoundsSink = Sinks.many().multicast().directBestEffort<LowerBoundData>()
+
    /**
     * Initiate a new head subscription with connection to the remote
     */
@@ -89,6 +93,10 @@ class GrpcHead(
                }

        var blocks = heads.map(converter)
+            .doOnNext {
+                it.lowerBounds.forEach { bound -> lowerBoundsSink.tryEmitNext(bound) }
+            }
+            .map { it.block }
            .distinctUntilChanged {
                it.hash
            }.filter { forkChoice.filter(it) }
@@ -125,8 +133,17 @@ class GrpcHead(
        headSubscription?.dispose()
    }

+    fun lowerBoundsFlux(): Flux<LowerBoundData> = lowerBoundsSink.asFlux()
+
    val headsCounter = Counter.builder("grpc_head_received")
        .tag("upstream", id)
        .tag("chain", chain.chainCode)
        .register(Metrics.globalRegistry)
+
+    data class GrpcHeadData(
+        val block: BlockContainer,
+        val lowerBounds: List<LowerBoundData>,
+    ) {
+        constructor(block: BlockContainer) : this(block, emptyList())
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamCreator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamCreator.kt
@@ -0,0 +1,72 @@
+package io.emeraldpay.dshackle.upstream.grpc
+
+import brave.grpc.GrpcTracing
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.FileResolver
+import io.emeraldpay.dshackle.config.AuthorizationConfig
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.CompressionConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.startup.configure.UpstreamCreator.Companion.getHash
+import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Component
+import reactor.core.scheduler.Scheduler
+import reactor.core.scheduler.Schedulers
+import java.util.concurrent.Executor
+import java.util.concurrent.Executors
+
+@Component
+class GrpcUpstreamCreator(
+    private val authorizationConfig: AuthorizationConfig,
+    private val compressionConfig: CompressionConfig,
+    private val fileResolver: FileResolver,
+    @Qualifier("grpcChannelExecutor")
+    private val channelExecutor: Executor,
+    private val grpcTracing: GrpcTracing,
+    @Qualifier("headScheduler")
+    private val headScheduler: Scheduler,
+    private val grpcAuthContext: GrpcAuthContext,
+) {
+    @Value("\${spring.application.max-metadata-size}")
+    private var maxMetadataSize: Int = Defaults.maxMetadataSize
+
+    private val hashes: MutableMap<Byte, Boolean> = HashMap()
+
+    companion object {
+        val grpcUpstreamsScheduler: Scheduler = Schedulers.fromExecutorService(
+            Executors.newFixedThreadPool(2),
+            "GrpcUpstreamsStatuses",
+        )
+    }
+
+    fun creatGrpcUpstream(
+        config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
+        chainsConfig: ChainsConfig,
+    ): GrpcUpstreams {
+        val endpoint = config.connection!!
+        return GrpcUpstreams(
+            config.id!!,
+            getHash(config.nodeId, "${endpoint.host}:${endpoint.port}", hashes),
+            config.role,
+            endpoint.host!!,
+            endpoint.port,
+            endpoint.auth,
+            endpoint.tokenAuth,
+            authorizationConfig,
+            compressionConfig.grpc.clientEnabled,
+            fileResolver,
+            endpoint.upstreamRating,
+            config.labels,
+            grpcUpstreamsScheduler,
+            channelExecutor,
+            chainsConfig,
+            grpcTracing,
+            null,
+            maxMetadataSize,
+            headScheduler,
+            grpcAuthContext,
+        )
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt
@@ -1,5 +1,6 @@
 package io.emeraldpay.dshackle.upstream.lowerbound

+import io.emeraldpay.api.proto.BlockchainOuterClass
 import java.time.Instant

 data class LowerBoundData(
@@ -21,3 +22,13 @@ data class LowerBoundData(
 enum class LowerBoundType {
    UNKNOWN, STATE, SLOT, BLOCK
 }
+
+fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType {
+    return when (this) {
+        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_SLOT -> LowerBoundType.SLOT
+        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_UNSPECIFIED -> LowerBoundType.UNKNOWN
+        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_STATE -> LowerBoundType.STATE
+        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOCK -> LowerBoundType.BLOCK
+        BlockchainOuterClass.LowerBoundType.UNRECOGNIZED -> LowerBoundType.UNKNOWN
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
@@ -29,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
 import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
 import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
+import io.emeraldpay.dshackle.upstream.stream.Chunk
 import io.grpc.StatusRuntimeException
 import org.apache.commons.lang3.time.StopWatch
 import reactor.core.publisher.Mono
@@ -53,6 +54,9 @@ class JsonRpcGrpcClient(
        override fun read(key: ChainRequest): Mono<ChainResponse> {
            val timer = StopWatch()
            val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
+                .setChunkSize(
+                    if (key.isStreamed) 500 else 0,
+                )
                .setChainValue(chain.id)

            key.selector?.let { req.selector = it }
@@ -91,9 +95,22 @@ class JsonRpcGrpcClient(
                .doOnNext { timer.start() }
                .flatMap {
                    stub.nativeCall(req.build())
+                        .switchOnFirst({ first, responseStream ->
+                            if (first.get()!!.chunked) {
+                                Mono.just(
+                                    ChainResponse(
+                                        responseStream.map { Chunk(it.payload.toByteArray(), it.finalChunk) },
+                                        key.id,
+                                    ),
+                                )
+                            } else {
+                                responseStream
+                                    .single()
+                                    .flatMap(::handleResponse)
+                            }
+                        }, false,)
                        .single()
                        .onErrorResume(::handleError)
-                        .flatMap(::handleResponse)
                }
                .doOnNext {
                    if (timer.isStarted) {
@@ -122,7 +139,7 @@ class JsonRpcGrpcClient(
                )
            }

-        fun handleError(t: Throwable): Mono<BlockchainOuterClass.NativeCallReplyItem> {
+        fun handleError(t: Throwable): Mono<ChainResponse> {
            metrics?.fails?.increment()
            return when (t) {
                is StatusRuntimeException -> Mono.error(