From 1ec3d444843d4beb2e288ebdc562bcb7ab45137d Mon Sep 17 00:00:00 2001 From: Vyacheslav Shebanov Date: Thu, 23 Feb 2023 20:25:46 +0200 Subject: [PATCH] =?UTF-8?q?new=20upstream=20change=20type=20=E2=80=94=20up?= =?UTF-8?q?dated,=20triggered=20when=20upstream=20methods=20changed=20(#14?= =?UTF-8?q?8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dshackle/startup/UpstreamChangeEvent.kt | 5 +++ .../dshackle/upstream/Multistream.kt | 43 +++++++++++-------- .../upstream/grpc/BitcoinGrpcUpstream.kt | 8 ++-- .../upstream/grpc/EthereumGrpcUpstream.kt | 8 ++-- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 8 ++-- .../dshackle/upstream/grpc/GrpcUpstream.kt | 2 +- .../upstream/grpc/GrpcUpstreamStatus.kt | 4 +- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 16 ++++--- .../grpc/GrpcUpstreamStatusSpec.groovy | 36 ++++++++++++++-- 9 files changed, 93 insertions(+), 37 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt index 1b8f35de..82bd9b5c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt @@ -45,6 +45,11 @@ class UpstreamChangeEvent( */ ADDED, + /** + * Some upstream details changed + */ + UPDATED, + /** * Upstream become available after being temporally off */ diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index b6d3c289..d258d427 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -361,28 +361,35 @@ abstract class Multistream( val chain = event.chain if (this.chain == chain) { eventLock.withLock { - if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) { - removeUpstream(event.upstream.getId()).takeIf { it }?.let { - try { - removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed") - } catch (e: Sinks.EmissionException) { - log.error("error during event processing $event", e) - } + when (event.type) { + UpstreamChangeEvent.ChangeType.REVALIDATED -> {} + UpstreamChangeEvent.ChangeType.UPDATED -> { + onUpstreamsUpdated() } - } else { - if (event.upstream is CachesEnabled) { - event.upstream.setCaches(caches) - } - addUpstream(event.upstream).takeIf { it }?.let { + UpstreamChangeEvent.ChangeType.ADDED -> { if (!started) { start() } - try { - addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - log.info("Upstream ${event.upstream.getId()} with chain $chain has been added") - } catch (e: Sinks.EmissionException) { - log.error("error during event processing $event", e) + if (event.upstream is CachesEnabled) { + event.upstream.setCaches(caches) + } + addUpstream(event.upstream).takeIf { it }?.let { + try { + addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + log.info("Upstream ${event.upstream.getId()} with chain $chain has been added") + } catch (e: Sinks.EmissionException) { + log.error("error during event processing $event", e) + } + } + } + UpstreamChangeEvent.ChangeType.REMOVED -> { + removeUpstream(event.upstream.getId()).takeIf { it }?.let { + try { + removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed") + } catch (e: Sinks.EmissionException) { + log.error("error during event processing $event", e) + } } } } 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 3e8db1d1..859168b0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -149,9 +149,11 @@ class BitcoinGrpcUpstream( override fun stop() { } - override fun update(conf: BlockchainOuterClass.DescribeChain) { - upstreamStatus.update(conf) - this.capabilities = RemoteCapabilities.extract(conf) + override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean { + val newCapabilities = RemoteCapabilities.extract(conf) conf.status?.let { status -> onStatus(status) } + return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also { + capabilities = newCapabilities + } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 1b2891b2..2b1fbbae 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -131,10 +131,12 @@ open class EthereumGrpcUpstream( override fun stop() { } - override fun update(conf: BlockchainOuterClass.DescribeChain) { - upstreamStatus.update(conf) - capabilities = RemoteCapabilities.extract(conf) + override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean { + val newCapabilities = RemoteCapabilities.extract(conf) conf.status?.let { status -> onStatus(status) } + return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also { + capabilities = newCapabilities + } } override fun getQuorumByLabel(): QuorumForLabels { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index f2ecd62b..23561a81 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -125,10 +125,12 @@ open class EthereumPosGrpcUpstream( override fun stop() { } - override fun update(conf: BlockchainOuterClass.DescribeChain) { - upstreamStatus.update(conf) - capabilities = RemoteCapabilities.extract(conf) + override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean { + val newCapabilities = RemoteCapabilities.extract(conf) conf.status?.let { status -> onStatus(status) } + return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also { + capabilities = newCapabilities + } } override fun getQuorumByLabel(): QuorumForLabels { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt index 883085c3..877d1c0e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -27,7 +27,7 @@ interface GrpcUpstream : Upstream { * Update the configuration of the upstream with the new data. * Called on the first creation, and each time a new state received from upstream */ - fun update(conf: BlockchainOuterClass.DescribeChain) + fun update(conf: BlockchainOuterClass.DescribeChain): Boolean fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub 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 335f97cc..7f9fe35d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt @@ -35,7 +35,7 @@ class GrpcUpstreamStatus( private val nodes = AtomicReference(QuorumForLabels()) private var targets: CallMethods? = null - fun update(conf: BlockchainOuterClass.DescribeChain) { + fun update(conf: BlockchainOuterClass.DescribeChain): Boolean { val updateLabels = ArrayList() val updateNodes = QuorumForLabels() @@ -55,7 +55,9 @@ class GrpcUpstreamStatus( this.nodes.set(updateNodes) this.allLabels.set(Collections.unmodifiableCollection(updateLabels)) + val changed = conf.supportedMethodsList.toSet() != this.targets?.getSupportedMethods() this.targets = DirectCallMethods(conf.supportedMethodsList.toSet()) + return changed } fun getLabels(): Collection { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 6b8f3abc..85e7e5ba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -149,8 +149,10 @@ class GrpcUpstreams( try { val chain = Chain.byId(chainDetails.chain.number) val up = getOrCreate(chain) - (up.upstream as GrpcUpstream).update(chainDetails) - up + val changed = (up.upstream as GrpcUpstream).update(chainDetails) + up.takeUnless { + changed && it.type == UpstreamChangeEvent.ChangeType.REVALIDATED + } ?: UpstreamChangeEvent(up.chain, up.upstream, UpstreamChangeEvent.ChangeType.UPDATED) } catch (e: Throwable) { log.warn("Skip unsupported upstream ${chainDetails.chain} on $id: ${e.message}") null @@ -161,16 +163,20 @@ class GrpcUpstreams( it.type == UpstreamChangeEvent.ChangeType.ADDED } + val updated = current.filter { + it.type == UpstreamChangeEvent.ChangeType.UPDATED + } + val removed = known.filterNot { kv -> val stillCurrent = current.any { c -> c.chain == kv.key } stillCurrent }.map { UpstreamChangeEvent(it.key, known.remove(it.key)!!, UpstreamChangeEvent.ChangeType.REMOVED) } - 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 }}") + if (removed.isNotEmpty() || added.isNotEmpty() || updated.isNotEmpty()) { + log.info("Finished processing of grpc upstream description for $id with content delta added ${added.map { it.chain }}, updated ${updated.map { it.chain }} and removed ${removed.map { it.chain }}") } - return Flux.fromIterable(removed + added) + return Flux.fromIterable(removed + added + updated) } private fun withTls(auth: AuthConfig.ClientTlsAuth): SslContext { 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 ad98480e..c0014a55 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy @@ -65,7 +65,7 @@ class GrpcUpstreamStatusSpec extends Specification { // more values when: - status.update( + def result = status.update( BlockchainOuterClass.DescribeChain.newBuilder() .addNodes( BlockchainOuterClass.NodeDetails.newBuilder() @@ -81,6 +81,7 @@ class GrpcUpstreamStatusSpec extends Specification { ) act = status.getLabels() then: + !result act.toList() == [ UpstreamsConfig.Labels.fromMap([test1: "bar", test2: "baz"]) ] @@ -109,7 +110,7 @@ class GrpcUpstreamStatusSpec extends Specification { // replace with new value when: - status.update( + def result = status.update( BlockchainOuterClass.DescribeChain.newBuilder() .addNodes( BlockchainOuterClass.NodeDetails.newBuilder() @@ -124,6 +125,7 @@ class GrpcUpstreamStatusSpec extends Specification { ) act = status.getLabels() then: + !result act.toList() == [ UpstreamsConfig.Labels.fromMap([test: "bar", fix: "value"]) ] @@ -202,7 +204,7 @@ class GrpcUpstreamStatusSpec extends Specification { setup: def status = new GrpcUpstreamStatus(null) when: - status.update( + def result = status.update( BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods([ "test_1", @@ -212,7 +214,35 @@ class GrpcUpstreamStatusSpec extends Specification { ) def act = status.getCallMethods() then: + result act.supportedMethods == ["test_1", "test_2"].toSet() act instanceof DirectCallMethods } + + def "Updates with new methods"() { + setup: + def status = new GrpcUpstreamStatus(null) + status.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .addAllSupportedMethods([ + "test_1", + "test_2" + ]) + .build() + ) + when: + def result = status.update( + BlockchainOuterClass.DescribeChain.newBuilder() + .addAllSupportedMethods([ + "test_1", + "test_2", + "test_3" + ]) + .build()) + def act = status.getCallMethods() + then: + result + act.supportedMethods == ["test_1", "test_2", "test_3"].toSet() + act instanceof DirectCallMethods + } }