diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 9fda9872..5f4e1585 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -19,9 +19,7 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.Capability -import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.MultistreamHolder import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -48,28 +46,21 @@ class Describe( .addAllSupportedSubscriptions(chainUpstreams.getEgressSubscription().getAvailableTopics()) .setStatus(status) .setCurrentHeight(chainUpstreams.getHead().getCurrentHeight() ?: 0) - chainUpstreams.getAll().let { ups -> - ups.forEach { up -> - val nodes = QuorumForLabels() - if (up is DefaultUpstream) { - nodes.add(up.getQuorumByLabel()) - } - nodes.getAll().forEach { node -> - val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder() - .setQuorum(node.quorum) - .addAllLabels( - node.labels.entries.map { label -> - BlockchainOuterClass.Label.newBuilder() - .setName(label.key) - .setValue(label.value) - .build() - } - ) - chainDescription.addNodes(nodeDetails) - } - capabilities.addAll(up.getCapabilities()) + chainUpstreams.getQuorumLabels() + .forEach { node -> + val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder() + .setQuorum(node.quorum) + .addAllLabels( + node.labels.entries.map { label -> + BlockchainOuterClass.Label.newBuilder() + .setName(label.key) + .setValue(label.value) + .build() + } + ) + chainDescription.addNodes(nodeDetails) } - } + capabilities.addAll(chainUpstreams.getCapabilities()) chainDescription.addAllCapabilities( capabilities.map { when (it) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 48582563..2d41711e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.JsonRpcReader +import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -63,6 +64,7 @@ abstract class Multistream( private var cacheSubscription: Disposable? = null private val reconfigLock = ReentrantLock() private val eventLock = ReentrantLock() + @Volatile private var callMethods: CallMethods? = null private var callMethodsFactory: Factory = Factory { return@Factory callMethods ?: throw FunctorException("Not initialized yet") @@ -70,7 +72,10 @@ abstract class Multistream( private var seq = 0 protected var lagObserver: HeadLagObserver? = null private var subscription: Disposable? = null + @Volatile private var capabilities: Set = emptySet() + @Volatile + private var quorumLabels: List? = null private val removed: MutableMap = HashMap() private val meters: MutableMap> = HashMap() private val addedUpstreams = Sinks.many() @@ -203,13 +208,14 @@ abstract class Multistream( open fun onUpstreamsUpdated() { reconfigLock.withLock { val upstreams = getAll() - upstreams.filter { it.isAvailable() }.map { it.getMethods() }.let { + val availableUpstreams = upstreams.filter { it.isAvailable() } + availableUpstreams.map { it.getMethods() }.let { callMethods = AggregatedCallMethods(it) } capabilities = if (upstreams.isEmpty()) { emptySet() } else { - upstreams.filter { it.isAvailable() }.map { up -> + availableUpstreams.map { up -> up.getCapabilities() }.let { if (it.isNotEmpty()) { @@ -219,6 +225,7 @@ abstract class Multistream( } } } + quorumLabels = getQuorumLabels(availableUpstreams) when { upstreams.size == 1 -> { lagObserver?.stop() @@ -230,6 +237,18 @@ abstract class Multistream( } } + private fun getQuorumLabels(ups: List): List { + val nodes = QuorumForLabels() + ups.forEach { up -> + if (up is DefaultUpstream) { + nodes.add(up.getQuorumByLabel()) + } + } + return nodes.getAll() + } + + fun getQuorumLabels(): List = quorumLabels ?: emptyList() + override fun observeStatus(): Flux { val upstreamsFluxes = getAll().map { up -> Flux.concat( @@ -457,9 +476,4 @@ abstract class Multistream( return map.values.min() } } - - data class UpstreamChangeState( - val upId: String, - val status: UpstreamAvailability - ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt index 7f9fe35d..2b269597 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt @@ -54,10 +54,11 @@ class GrpcUpstreamStatus( } this.nodes.set(updateNodes) + val labelsChanged = updateLabels != this.allLabels.get().toList() this.allLabels.set(Collections.unmodifiableCollection(updateLabels)) val changed = conf.supportedMethodsList.toSet() != this.targets?.getSupportedMethods() this.targets = DirectCallMethods(conf.supportedMethodsList.toSet()) - return changed + return changed || labelsChanged } fun getLabels(): Collection { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy index c0014a55..c588a06a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy @@ -81,7 +81,7 @@ class GrpcUpstreamStatusSpec extends Specification { ) act = status.getLabels() then: - !result + result act.toList() == [ UpstreamsConfig.Labels.fromMap([test1: "bar", test2: "baz"]) ] @@ -125,7 +125,7 @@ class GrpcUpstreamStatusSpec extends Specification { ) act = status.getLabels() then: - !result + result act.toList() == [ UpstreamsConfig.Labels.fromMap([test: "bar", fix: "value"]) ] diff --git a/src/test/kotlin/io/emeraldpay/dshackle/IntegrationTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/IntegrationTest.kt index 01350b17..e2208759 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/IntegrationTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/IntegrationTest.kt @@ -72,7 +72,6 @@ class IntegrationTest { val result = stub.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) Assertions.assertThat(result.chainsCount).isEqualTo(1) Assertions.assertThat(result.chainsList[0].chain).isEqualTo(ChainRef.CHAIN_ETHEREUM__MAINNET) - Assertions.assertThat(result.chainsList[0].nodesCount).isEqualTo(1) } @TestConfiguration