From 71d68e858be2b734e9e3bc832b0a840a46dd38da Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Thu, 30 May 2024 14:03:53 +0400 Subject: [PATCH] Reanimate grpc upstreams (#491) --- .../config/context/SchedulersConfig.kt | 6 ++ .../dshackle/startup/ConfiguredUpstreams.kt | 15 ++++ .../configure/GenericUpstreamCreator.kt | 25 +------ .../startup/configure/UpstreamCreator.kt | 25 +++++++ .../startup/configure/UpstreamFactory.kt | 11 +++ .../dshackle/upstream/DefaultUpstream.kt | 11 ++- .../emeraldpay/dshackle/upstream/Selector.kt | 14 +--- .../upstream/bitcoin/BitcoinUpstream.kt | 2 +- .../upstream/generic/GenericMultistream.kt | 2 +- .../upstream/generic/GenericUpstream.kt | 9 +-- .../upstream/grpc/BitcoinGrpcUpstream.kt | 4 +- .../upstream/grpc/GenericGrpcUpstream.kt | 37 ++++++++-- .../dshackle/upstream/grpc/GrpcHead.kt | 19 ++++- .../upstream/grpc/GrpcUpstreamCreator.kt | 72 +++++++++++++++++++ .../upstream/lowerbound/LowerBoundData.kt | 11 +++ .../upstream/rpcclient/JsonRpcGrpcClient.kt | 21 +++++- .../upstream/grpc/GrpcHeadSpec.groovy | 4 +- .../startup/configure/UpstreamFactoryTest.kt | 9 ++- 18 files changed, 237 insertions(+), 60 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamCreator.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt index cf15afce..a174b0d6 100644 --- 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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index d4367ad2..11a2fff5 100644 --- 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, + 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) + } } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt index f3c4e9fb..26f9ce28 100644 --- 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 { 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() - } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt index 52932cf2..1dcfe51a 100644 --- 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 = + nodeId?.toByte() ?: (obj.hashCode() % 255).let { + if (it == 0) 1 else it + }.let { nonZeroHash -> + listOf>( + 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, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt index 465f244b..9bef5c11 100644 --- 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, + chainsConfig: ChainsConfig, + ): GrpcUpstreams { + return grpcUpstreamCreator.creatGrpcUpstream(config, chainsConfig) + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 27ee20f2..6def8615 100644 --- 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) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index e5d63271..a49a2dd5 100644 --- 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt index 5e9f3e7a..2f9cdf9d 100644 --- 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, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt index c6e5703f..564ae4c6 100644 --- 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( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt index f65d96c5..33756867 100644 --- 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 } - } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index c6e0139c..42c4b5e1 100644 --- 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 = Function { value -> + private val blockConverter: Function = 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> = Function { existingBlock -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt index c1a2331e..73633b00 100644 --- 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 = Function { value -> + private val blockConverter: Function = 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() + 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 { - 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-")), + ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index 4fac0058..b0581a50 100644 --- 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, + private val converter: Function, /** * 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() + /** * 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 = 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, + ) { + constructor(block: BlockContainer) : this(block, emptyList()) + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamCreator.kt new file mode 100644 index 00000000..1a96a75b --- /dev/null +++ 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 = HashMap() + + companion object { + val grpcUpstreamsScheduler: Scheduler = Schedulers.fromExecutorService( + Executors.newFixedThreadPool(2), + "GrpcUpstreamsStatuses", + ) + } + + fun creatGrpcUpstream( + config: UpstreamsConfig.Upstream, + 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, + ) + } +} 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 47342b66..2ce89f64 100644 --- 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 + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt index 22f1c6d4..d67d26af 100644 --- 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 { 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 { + fun handleError(t: Throwable): Mono { metrics?.fails?.increment() return when (t) { is StatusRuntimeException -> Mono.error( diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index cfb375a4..36ba9445 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy @@ -60,7 +60,7 @@ class GrpcHeadSpec extends Specification { } }) def convert = { BlockchainOuterClass.ChainHead head -> - TestingCommons.blockForBitcoin(head.height) + new GrpcHead.GrpcHeadData(TestingCommons.blockForBitcoin(head.height), List.of()) } def head = new GrpcHead( "test", @@ -127,7 +127,7 @@ class GrpcHeadSpec extends Specification { } }) def convert = { BlockchainOuterClass.ChainHead head -> - TestingCommons.blockForBitcoin(head.height) + new GrpcHead.GrpcHeadData(TestingCommons.blockForBitcoin(head.height), List.of()) } def head = new GrpcHead( "test", diff --git a/src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt index eecccb8c..c751c35c 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.startup.configure import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreamCreator import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.Arguments @@ -33,7 +34,13 @@ class UpstreamFactoryTest { private val genericUpstreamCreator = mock() private val ethereumUpstreamCreator = mock() private val bitcoinUpstreamCreator = mock() - private val upstreamFactory = UpstreamFactory(genericUpstreamCreator, ethereumUpstreamCreator, bitcoinUpstreamCreator) + private val grpcUpstreamCreator = mock() + private val upstreamFactory = UpstreamFactory( + genericUpstreamCreator, + ethereumUpstreamCreator, + bitcoinUpstreamCreator, + grpcUpstreamCreator, + ) @JvmStatic fun data() = listOf(