Merge pull request #110 from p2p-org/stable-multistreams

Stabilisation of upstreams

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

@@ -22,33 +22,30 @@ import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
-import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 
 @Service
 class SubscribeStatus(
-    @Autowired private val multistreamHolder: MultistreamHolder
+    private val multistreamHolder: MultistreamHolder
 ) {
 
     fun subscribeStatus(requestMono: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
         return requestMono.flatMapMany { req ->
-            // check status for all requested chains
+            if (req.chainsCount == 0) {
-            val all = req.chainsList.map {
+                Flux.error(RuntimeException("empty chains list"))
-                val chain = Chain.byId(it.number)
+            } else {
-                val up = multistreamHolder.getUpstream(chain)
+                // check status for all requested chains
-                if (up == null) {
+                val all = req.chainsList.map {
-                    // when the chain is not configured return just UNAVAILABLE
+                    val chain = Chain.byId(it.number)
-                    Mono.just(chainUnavailable(chain)).flux()
+                    val up = multistreamHolder.getUpstream(chain)
-                } else {
-                    // when configured subscribe to its updates
                     up.observeStatus().map { availability ->
                         chainStatus(chain, availability, up)
                     }
                 }
+                Flux.merge(all)
             }
-            Flux.merge(all)
         }
     }
 
@@ -74,6 +71,4 @@ class SubscribeStatus(
             .setQuorum(quorum)
             .build()
     }
-
-    class ChainSubscription(val chain: Chain, val up: Multistream, val avail: UpstreamAvailability)
 }

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -47,11 +47,11 @@ import org.springframework.boot.ApplicationArguments
 import org.springframework.boot.ApplicationRunner
 import org.springframework.context.ApplicationEventPublisher
 import org.springframework.stereotype.Component
-import reactor.core.publisher.Flux
+import reactor.core.scheduler.Scheduler
-import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 import java.net.URI
 import java.util.concurrent.Executor
+import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.function.Function
 import kotlin.math.abs
@@ -68,13 +68,13 @@ open class ConfiguredUpstreams(
 
     private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
     private var seq = AtomicInteger(0)
-
     private val hashes: MutableMap<Byte, Boolean> = HashMap()
 
+    lateinit var grpcUpstreamsScheduler: Scheduler
+
     override fun run(args: ApplicationArguments) {
         log.debug("Starting upstreams")
         val defaultOptions = buildDefaultOptions(config)
-        val observerScheduler = Schedulers.newParallel("status-observer", 3)
         config.upstreams.forEach { up ->
             if (!up.isEnabled) {
                 log.debug("Upstream ${up.id} is disabled")
@@ -111,20 +111,8 @@ open class ConfiguredUpstreams(
                     }
                 }
                 upstream?.let {
-                    Flux.concat(Mono.just(UpstreamChangeEvent.ChangeType.ADDED), upstream.observeStatus())
+                    val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED)
-                        .distinctUntilChanged()
+                    eventPublisher.publishEvent(event)
-                        .subscribeOn(observerScheduler)
-                        .subscribe { status ->
-                            when (status) {
-                                UpstreamAvailability.UNAVAILABLE -> UpstreamChangeEvent.ChangeType.REMOVED
-                                else -> UpstreamChangeEvent.ChangeType.REVALIDATED
-                            }.let { eventType ->
-                                if (eventType == UpstreamChangeEvent.ChangeType.REMOVED) {
-                                    log.warn("Remove upstream ${it::class.java.simpleName}:${upstream.getId()} due to $status")
-                                }
-                                eventPublisher.publishEvent(UpstreamChangeEvent(chain, upstream, eventType))
-                            }
-                        }
                 }
             }
         }
@@ -301,6 +289,12 @@ open class ConfiguredUpstreams(
         config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
         options: UpstreamsConfig.Options
     ) {
+        if (!this::grpcUpstreamsScheduler.isInitialized) {
+            grpcUpstreamsScheduler = Schedulers.fromExecutorService(
+                Executors.newFixedThreadPool(2),
+                "GrpcUpstreamsStatuses"
+            )
+        }
         val endpoint = config.connection!!
         val hash = getHash(nodeId, "${endpoint.host}:${endpoint.port}")
         val ds = GrpcUpstreams(
@@ -313,6 +307,7 @@ open class ConfiguredUpstreams(
             fileResolver,
             endpoint.upstreamRating,
             config.labels,
+            grpcUpstreamsScheduler,
             channelExecutor
         ).apply {
             timeout = options.timeout

src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt

@@ -80,7 +80,6 @@ abstract class DefaultUpstream(
 
     fun onStatus(value: BlockchainOuterClass.ChainStatus) {
         val available = value.availability
-        val quorum = value.quorum
         setStatus(
             if (available != null) UpstreamAvailability.fromGrpc(available.number) else UpstreamAvailability.UNAVAILABLE
         )
@@ -98,7 +97,7 @@ abstract class DefaultUpstream(
         }
     }
 
-    fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
+    private fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
         if (options.disableValidation == true) {
             // if we specifically told that this upstream should be _always valid_ then skip
             // the status calculation and trust the proposed value as is

src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt

@@ -34,6 +34,7 @@ open class DynamicMergedHead(
 
     override fun stop() {
         super.stop()
+        log.info("Stop dynamic merged head for [$label]")
         dynamicFlux.stop()
         subscription?.dispose()
         subscription = null

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

@@ -48,7 +48,7 @@ open class EthereumMultistream(
 
     private var head: DynamicMergedHead = DynamicMergedHead(
         PriorityForkChoice(),
-        "ETH Multistream",
+        "ETH Multistream of ${chain.chainCode}",
         headScheduler
     )
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -47,7 +47,7 @@ open class EthereumPosMultiStream(
 
     private var head: DynamicMergedHead = DynamicMergedHead(
         PriorityForkChoice(),
-        "ETH Pos Multistream",
+        "ETH Pos Multistream of ${chain.chainCode}",
         headScheduler
     )
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt

@@ -98,7 +98,7 @@ class BitcoinGrpcUpstream(
     }
     private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
     private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
-    var timeout = Defaults.timeout
+    private val timeout = Defaults.timeout
     private var capabilities: Set<Capability> = emptySet()
 
     override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt

@@ -108,7 +108,7 @@ open class EthereumPosGrpcUpstream(
     private var capabilities: Set<Capability> = emptySet()
 
     private val defaultReader: JsonRpcReader = client.getReader()
-    var timeout = Defaults.timeout
+    private val timeout = Defaults.timeout
     private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
 
     override fun start() {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt

@@ -16,7 +16,8 @@
  */
 package io.emeraldpay.dshackle.upstream.grpc
 
-import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.BlockchainOuterClass.*
+import io.emeraldpay.api.proto.Common
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.Chain
@@ -26,6 +27,7 @@ import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
@@ -43,6 +45,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils
 import org.slf4j.LoggerFactory
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
+import reactor.core.scheduler.Scheduler
 import java.io.IOException
 import java.time.Duration
 import java.util.concurrent.Executor
@@ -60,13 +63,14 @@ class GrpcUpstreams(
     private val fileResolver: FileResolver,
     private val nodeRating: Int,
     private val labels: UpstreamsConfig.Labels,
-    private val executor: Executor
+    private val chainStatusScheduler: Scheduler,
+    private val grpcExecutor: Executor
 ) {
     private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
 
     var timeout = Defaults.timeout
 
-    private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null
+    lateinit var client: ReactorBlockchainGrpc.ReactorBlockchainStub
     private val known = HashMap<Chain, DefaultUpstream>()
     private val lock = ReentrantLock()
 
@@ -75,7 +79,7 @@ class GrpcUpstreams(
             // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces)
             .maxInboundMessageSize(Defaults.maxMessageSize)
             .enableRetry()
-            .executor(executor)
+            .executor(grpcExecutor)
             .maxRetryAttempts(3)
         if (auth != null && StringUtils.isNotEmpty(auth.ca)) {
             chanelBuilder
@@ -91,9 +95,9 @@ class GrpcUpstreams(
 
         val statusSubscription = AtomicReference<Disposable>()
 
-        val updates = Flux.interval(Duration.ZERO, Duration.ofSeconds(20))
+        return Flux.interval(Duration.ZERO, Duration.ofSeconds(20))
             .flatMap {
-                client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
+                client.describe(DescribeRequest.newBuilder().build())
             }.onErrorContinue { t, _ ->
                 if (ExceptionUtils.indexOfType(t, IOException::class.java) >= 0) {
                     log.warn("gRPC upstream $host:$port is unavailable. (${t.javaClass}: ${t.message})")
@@ -106,7 +110,10 @@ class GrpcUpstreams(
             }.flatMap { value ->
                 processDescription(value)
             }.doOnNext {
-                val subscription = client.subscribeStatus(BlockchainOuterClass.StatusRequest.newBuilder().build())
+                val subscription = client.subscribeStatus(
+                    StatusRequest.newBuilder()
+                        .addChains(Common.ChainRef.forNumber(it.chain.id)).build()
+                ).subscribeOn(chainStatusScheduler)
                     .subscribe { value ->
                         val chain = Chain.byId(value.chain.number)
                         if (chain != Chain.UNSPECIFIED) {
@@ -120,11 +127,9 @@ class GrpcUpstreams(
             }.doOnError { t ->
                 log.error("Failed to process update from gRPC upstream $id", t)
             }
-
-        return updates
     }
 
-    fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux<UpstreamChangeEvent> {
+    private fun processDescription(value: DescribeResponse): Flux<UpstreamChangeEvent> {
         log.info("Start processing grpc upstream description for $id with chains ${value.chainsList.map { it.chain.name }}")
         val current = value.chainsList.filter {
             Chain.byId(it.chain.number) != Chain.UNSPECIFIED
@@ -150,11 +155,13 @@ class GrpcUpstreams(
         }.map {
             UpstreamChangeEvent(it.key, known.remove(it.key)!!, UpstreamChangeEvent.ChangeType.REMOVED)
         }
-        log.info("Finished processing of grpc upstream description for $id with content delta added [${added.map { it.chain }}] and removed [${removed.map { it.chain }}]")
+        if (removed.isNotEmpty() || added.isNotEmpty()) {
+            log.info("Finished processing of grpc upstream description for $id with content delta added ${added.map { it.chain }} and removed ${removed.map { it.chain }}")
+        }
         return Flux.fromIterable(removed + added)
     }
 
-    internal fun withTls(auth: AuthConfig.ClientTlsAuth): SslContext {
+    private fun withTls(auth: AuthConfig.ClientTlsAuth): SslContext {
         val sslContext = SslContextBuilder.forClient()
             .clientAuth(ClientAuth.REQUIRE)
         sslContext.trustManager(fileResolver.resolve(auth.ca!!).inputStream())
@@ -176,13 +183,31 @@ class GrpcUpstreams(
         return sslContext.build()
     }
 
-    fun getOrCreate(chain: Chain): UpstreamChangeEvent {
+    private val creators: Map<BlockchainType, (chain: Chain, client: JsonRpcGrpcClient) -> DefaultUpstream> = mapOf(
+        BlockchainType.EVM_POW to { chain, rpcClient ->
+            EthereumGrpcUpstream(id, hash, role, chain, client, rpcClient, labels)
+        },
+        BlockchainType.EVM_POS to { chain, rpcClient ->
+            EthereumPosGrpcUpstream(id, hash, role, chain, client, rpcClient, nodeRating, labels)
+        },
+        BlockchainType.BITCOIN to { chain, rpcClient ->
+            BitcoinGrpcUpstream(id, role, chain, client, rpcClient, labels)
+        }
+    )
+
+    private fun getOrCreate(chain: Chain): UpstreamChangeEvent {
+        val metrics = makeMetrics(chain)
+        val creator = creators.getValue(BlockchainType.from(chain))
+        return getOrCreate(chain, metrics, creator)
+    }
+
+    private fun makeMetrics(chain: Chain): RpcMetrics {
         val metricsTags = listOf(
             Tag.of("upstream", id),
             Tag.of("chain", chain.chainCode)
         )
 
-        val metrics = RpcMetrics(
+        return RpcMetrics(
             Timer.builder("upstream.grpc.conn")
                 .description("Request time through a Dshackle/gRPC connection")
                 .tags(metricsTags)
@@ -193,68 +218,24 @@ class GrpcUpstreams(
                 .tags(metricsTags)
                 .register(Metrics.globalRegistry)
         )
-
-        val blockchainType = BlockchainType.from(chain)
-        if (blockchainType == BlockchainType.EVM_POW) {
-            return getOrCreateEthereum(chain, metrics)
-        } else if (blockchainType == BlockchainType.BITCOIN) {
-            return getOrCreateBitcoin(chain, metrics)
-        } else if (blockchainType == BlockchainType.EVM_POS) {
-            return getOrCreateEthereumPos(chain, metrics)
-        } else {
-            throw IllegalArgumentException("Unsupported blockchain: $chain")
-        }
     }
 
-    fun getOrCreateEthereum(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent {
+    private fun getOrCreate(
+        chain: Chain,
+        metrics: RpcMetrics,
+        creator: (chain: Chain, client: JsonRpcGrpcClient) -> DefaultUpstream
+    ): UpstreamChangeEvent {
         lock.withLock {
             val current = known[chain]
             return if (current == null) {
-                val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
+                val rpcClient = JsonRpcGrpcClient(client, chain, metrics)
-                val created = EthereumGrpcUpstream(id, hash, role, chain, client!!, rpcClient, labels)
+                val created = creator(chain, rpcClient)
-                created.timeout = this.timeout
                 known[chain] = created
-                created.start()
+                if (created is Lifecycle) created.start()
                 UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED)
             } else {
                 UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED)
             }
         }
     }
-
-    fun getOrCreateEthereumPos(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent {
-        lock.withLock {
-            val current = known[chain]
-            return if (current == null) {
-                val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
-                val created = EthereumPosGrpcUpstream(id, hash, role, chain, client!!, rpcClient, nodeRating, labels)
-                created.timeout = this.timeout
-                known[chain] = created
-                created.start()
-                UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED)
-            } else {
-                UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED)
-            }
-        }
-    }
-
-    fun getOrCreateBitcoin(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent {
-        lock.withLock {
-            val current = known[chain]
-            return if (current == null) {
-                val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
-                val created = BitcoinGrpcUpstream(id, role, chain, client!!, rpcClient, labels)
-                created.timeout = this.timeout
-                known[chain] = created
-                created.start()
-                UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED)
-            } else {
-                UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED)
-            }
-        }
-    }
-
-    fun get(chain: Chain): DefaultUpstream {
-        return known[chain]!!
-    }
 }

src/test/groovy/io/emeraldpay/dshackle/rpc/SubscribeStatusSpec.groovy

@@ -15,7 +15,7 @@ import java.time.Duration
 
 class SubscribeStatusSpec extends Specification {
 
-    def "gives UNAVAIL if non-configured chain is requested"() {
+    def "returns requested statuses"() {
         setup:
         def ethereumUp = Mock(Upstream) {
             _ * getStatus() >> UpstreamAvailability.OK
@@ -28,17 +28,15 @@ class SubscribeStatusSpec extends Specification {
         def ups = Mock(MultistreamHolder) {
             _ * it.getAvailable() >> [Chain.ETHEREUM]
             1 * it.getUpstream(Chain.ETHEREUM) >> ethereumUpAll
-            1 * it.getUpstream(Chain.BITCOIN) >> null
         }
         def ctrl = new SubscribeStatus(ups)
 
         when:
         def req = BlockchainOuterClass.StatusRequest.newBuilder()
                 .addChains(Common.ChainRef.CHAIN_ETHEREUM)
-                .addChains(Common.ChainRef.CHAIN_BITCOIN)
                 .build()
         def act = ctrl.subscribeStatus(Mono.just(req))
-                .take(2)
+                .take(1)
         // sort just for testing
                 .sort(new Comparator<BlockchainOuterClass.ChainStatus>() {
                     @Override
@@ -48,9 +46,6 @@ class SubscribeStatusSpec extends Specification {
                 })
         then:
         StepVerifier.create(act)
-                .expectNextMatches {
-                    it.chainValue == Chain.BITCOIN.id && it.availability == BlockchainOuterClass.AvailabilityEnum.AVAIL_UNAVAILABLE
-                }
                 .expectNextMatches {
                     it.chainValue == Chain.ETHEREUM.id && it.availability == BlockchainOuterClass.AvailabilityEnum.AVAIL_OK
                 }