new upstream change type — updated, triggered when upstream methods changed (#148)

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
          */

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) {
+                when (event.type) {
-                    removeUpstream(event.upstream.getId()).takeIf { it }?.let {
+                    UpstreamChangeEvent.ChangeType.REVALIDATED -> {}
-                        try {
+                    UpstreamChangeEvent.ChangeType.UPDATED -> {
-                            removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                        onUpstreamsUpdated()
-                            log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
-                        } catch (e: Sinks.EmissionException) {
-                            log.error("error during event processing $event", e)
-                        }
                     }
-                } else {
+                    UpstreamChangeEvent.ChangeType.ADDED -> {
-                    if (event.upstream is CachesEnabled) {
-                        event.upstream.setCaches(caches)
-                    }
-                    addUpstream(event.upstream).takeIf { it }?.let {
                         if (!started) {
                             start()
                         }
-                        try {
+                        if (event.upstream is CachesEnabled) {
-                            addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                            event.upstream.setCaches(caches)
-                            log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
+                        }
-                        } catch (e: Sinks.EmissionException) {
+                        addUpstream(event.upstream).takeIf { it }?.let {
-                            log.error("error during event processing $event", e)
+                            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)
+                            }
                         }
                     }
                 }

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) {
+    override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean {
-        upstreamStatus.update(conf)
+        val newCapabilities = RemoteCapabilities.extract(conf)
-        this.capabilities = RemoteCapabilities.extract(conf)
         conf.status?.let { status -> onStatus(status) }
+        return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
+            capabilities = newCapabilities
+        }
     }
 }

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) {
+    override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean {
-        upstreamStatus.update(conf)
+        val newCapabilities = RemoteCapabilities.extract(conf)
-        capabilities = RemoteCapabilities.extract(conf)
         conf.status?.let { status -> onStatus(status) }
+        return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
+            capabilities = newCapabilities
+        }
     }
 
     override fun getQuorumByLabel(): QuorumForLabels {

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) {
+    override fun update(conf: BlockchainOuterClass.DescribeChain): Boolean {
-        upstreamStatus.update(conf)
+        val newCapabilities = RemoteCapabilities.extract(conf)
-        capabilities = RemoteCapabilities.extract(conf)
         conf.status?.let { status -> onStatus(status) }
+        return (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also {
+            capabilities = newCapabilities
+        }
     }
 
     override fun getQuorumByLabel(): QuorumForLabels {

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
 

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<UpstreamsConfig.Labels>()
         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<UpstreamsConfig.Labels> {

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)
+                val changed = (up.upstream as GrpcUpstream).update(chainDetails)
-                up
+                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()) {
+        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 }} and removed ${removed.map { it.chain }}")
+            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 {

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
+    }
 }