problem: grpc upstream can change set of provided upstreams

solution: periodically recheck and update current list

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -50,55 +50,72 @@ class UpstreamsConfigReader {
         }
 
         config.upstreams = ArrayList<UpstreamsConfig.Upstream<*>>()
-        getList<MappingNode>(configNode, "upstreams")?.value?.forEach { upNode ->
+        getList<MappingNode>(configNode, "upstreams")?.value?.forEachIndexed { pos, upNode ->
             val connNode = getMapping(upNode, "connection")
             if (hasAny(connNode, "ethereum")) {
                 val connConfigNode = getMapping(connNode, "ethereum")!!
                 val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>()
                 readUpstreamCommon(upNode, upstream)
                 readUpstreamEthereum(upNode, upstream)
-                config.upstreams.add(upstream)
+                if (isValid(upstream)) {
-                val connection = UpstreamsConfig.EthereumConnection()
+                    config.upstreams.add(upstream)
-                upstream.connection = connection
+                    val connection = UpstreamsConfig.EthereumConnection()
-                getMapping(connConfigNode, "rpc")?.let { node ->
+                    upstream.connection = connection
-                    getValueAsString(node, "url")?.let { url ->
+                    getMapping(connConfigNode, "rpc")?.let { node ->
-                        val http = UpstreamsConfig.HttpEndpoint(URI(url))
+                        getValueAsString(node, "url")?.let { url ->
-                        connection.rpc = http
+                            val http = UpstreamsConfig.HttpEndpoint(URI(url))
-                        http.basicAuth = readBasicAuth(node)
+                            connection.rpc = http
-                        http.tls = readTls(node)
+                            http.basicAuth = readBasicAuth(node)
-                    }
+                            http.tls = readTls(node)
-                }
-                getMapping(connConfigNode, "ws")?.let { node ->
-                    getValueAsString(node, "url")?.let { url ->
-                        val ws = UpstreamsConfig.WsEndpoint(URI(url))
-                        connection.ws = ws
-                        getValueAsString(node, "origin")?.let { origin ->
-                            ws.origin = URI(origin)
                         }
-                        ws.basicAuth = readBasicAuth(node)
                     }
+                    getMapping(connConfigNode, "ws")?.let { node ->
+                        getValueAsString(node, "url")?.let { url ->
+                            val ws = UpstreamsConfig.WsEndpoint(URI(url))
+                            connection.ws = ws
+                            getValueAsString(node, "origin")?.let { origin ->
+                                ws.origin = URI(origin)
+                            }
+                            ws.basicAuth = readBasicAuth(node)
+                        }
+                    }
+                } else {
+                    log.error("Upstream at #0 has invalid configuration")
                 }
             } else if (hasAny(connNode, "grpc")) {
                 val connConfigNode = getMapping(connNode, "grpc")!!
                 val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>()
                 readUpstreamCommon(upNode, upstream)
                 readUpstreamGrpc(upNode, upstream)
-                config.upstreams.add(upstream)
+                if (isValid(upstream)) {
-                val connection = UpstreamsConfig.GrpcConnection()
+                    config.upstreams.add(upstream)
-                upstream.connection = connection
+                    val connection = UpstreamsConfig.GrpcConnection()
-                getValueAsString(connConfigNode, "host")?.let {
+                    upstream.connection = connection
-                    connection.host = it
+                    getValueAsString(connConfigNode, "host")?.let {
+                        connection.host = it
+                    }
+                    getValueAsInt(connConfigNode, "port")?.let {
+                        connection.port = it
+                    }
+                    connection.auth = readTls(connConfigNode)
+                } else {
+                    log.error("Upstream at #0 has invalid configuration")
                 }
-                getValueAsInt(connConfigNode, "port")?.let {
-                    connection.port = it
-                }
-                connection.auth = readTls(connConfigNode)
             }
         }
 
         return config
     }
 
+    fun isValid(upstream: UpstreamsConfig.Upstream<*>): Boolean {
+        val id = upstream.id
+        if (id == null || id.length < 3 || !id.matches(Regex("[a-zA-Z][a-zA-Z0-9_-]+[a-zA-Z0-9]"))) {
+            log.warn("Invalid id: $id")
+            return false
+        }
+        return true
+    }
+
     internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
         upstream.id = getValueAsString(upNode, "id")
         upstream.options = tryReadOptions(upNode)

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

@@ -47,13 +47,13 @@ abstract class AggregatedUpstream(
     )
     var cache: CachingEthereumApi = CachingEthereumApi.empty()
     private val reconfigLock = ReentrantLock()
-    private var callMethods: CallMethods = DirectCallMethods()
+    private var callMethods: CallMethods? = null
 
     abstract fun getAll(): List<Upstream>
     abstract fun addUpstream(upstream: Upstream)
     abstract fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi>
 
-    fun reconfigure() {
+    fun onUpstreamsUpdated() {
         reconfigLock.withLock {
             getAll().map { it.getMethods() }.let {
                 callMethods = AggregatedCallMethods(it)
@@ -83,7 +83,7 @@ abstract class AggregatedUpstream(
     }
 
     override fun getMethods(): CallMethods {
-        return callMethods
+        return callMethods ?: throw IllegalStateException("Methods are not initialized yet")
     }
 
     class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())

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

@@ -35,12 +35,19 @@ open class ChainUpstreams (
 
     private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
     private var seq = 0
-    private var head: EthereumHead?
+    private var head: EthereumHead? = null
     private var lagObserver: HeadLagObserver? = null
     private var subscription: Disposable? = null
 
     init {
-        head = updateHead()
+        if (upstreams.size > 0) {
+            head = updateHead()
+            onUpstreamsUpdated()
+        }
+    }
+
+    override fun getId(): String {
+        return "!all:${chain.chainCode}"
     }
 
     override fun isRunning(): Boolean {
@@ -99,7 +106,14 @@ open class ChainUpstreams (
     override fun addUpstream(upstream: Upstream) {
         upstreams.add(upstream)
         head = updateHead()
-        reconfigure()
+        onUpstreamsUpdated()
+    }
+
+    fun removeUpstream(id: String) {
+        if (upstreams.removeIf { it.getId() == id }) {
+            head = updateHead()
+            onUpstreamsUpdated()
+        }
     }
 
     override fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi> {

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

@@ -43,13 +43,11 @@ import kotlin.collections.HashMap
 @Repository
 open class ConfiguredUpstreams(
         @Autowired val env: Environment,
-        @Autowired private val objectMapper: ObjectMapper
+        @Autowired private val objectMapper: ObjectMapper,
-) : Upstreams {
+        @Autowired private val currentUpstreams: CurrentUpstreams
+) {
 
     private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
-    private val chainMapping = ConcurrentHashMap<Chain, ChainUpstreams>()
-    private val chainsBus = TopicProcessor.create<Chain>()
-    private val callTargets = HashMap<Chain, QuorumBasedMethods>()
 
     private val chainNames = mapOf(
             "ethereum" to Chain.ETHEREUM,
@@ -67,7 +65,7 @@ open class ConfiguredUpstreams(
         config.upstreams.forEach { up ->
 
             if (up.connection is UpstreamsConfig.GrpcConnection) {
-                buildGrpcUpstream(up.connection as UpstreamsConfig.GrpcConnection)
+                buildGrpcUpstream(up as UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>)
             } else {
                 val chain = chainNames[up.chain]
                 if (chain == null) {
@@ -127,12 +125,12 @@ open class ConfiguredUpstreams(
         var rpcApi: DirectEthereumApi? = null
         val urls = ArrayList<URI>()
         val methods = if (config.methods != null) {
-            ManagedCallMethods(getDefaultMethods(chain),
+            ManagedCallMethods(currentUpstreams.getDefaultMethods(chain),
                     config.methods!!.enabled.map { it.name }.toSet(),
                     config.methods!!.disabled.map { it.name }.toSet()
             )
         } else {
-            getDefaultMethods(chain)
+            currentUpstreams.getDefaultMethods(chain)
         }
         conn.rpc?.let { endpoint ->
             val rpcTransport = DefaultRpcTransport(endpoint.url)
@@ -168,75 +166,32 @@ open class ConfiguredUpstreams(
             }
 
             log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
-            val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options,
+            val ethereumUpstream = EthereumUpstream(
+                    config.id!!,
+                    chain, rpcApi!!, wsApi, options,
                     NodeDetailsList.NodeDetails(1, config.labels),
                     methods)
             ethereumUpstream.start()
-            addUpstream(chain, ethereumUpstream)
+            currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED))
         }
     }
 
-    private fun buildGrpcUpstream(up: UpstreamsConfig.GrpcConnection) {
+    private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>) {
-        val endpoint = up
+        val endpoint = config.connection!!
-            val ds = GrpcUpstreams(
+        val ds = GrpcUpstreams(
-                    endpoint.host!!,
+                config.id!!,
-                    endpoint.port ?: 443,
+                endpoint.host!!,
-                    objectMapper,
+                endpoint.port ?: 443,
-                    up.auth
+                objectMapper,
-            )
+                endpoint.auth
-            log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
-            ds.start()
-                    .subscribe {
-                        log.info("Subscribed to ${it.t1} through gRPC at ${endpoint.host}:${endpoint.port}")
-                        addUpstream(it.t1, it.t2)
-                    }
-    }
-
-    override fun getUpstream(chain: Chain): AggregatedUpstream? {
-        return chainMapping[chain]
-    }
-
-    override fun addUpstream(chain: Chain, up: Upstream): ChainUpstreams {
-        val current = chainMapping[chain]
-        if (current == null) {
-            val created = ChainUpstreams(chain, ArrayList<Upstream>(), objectMapper)
-            created.addUpstream(up)
-            created.start()
-            chainMapping[chain] = created
-            chainsBus.onNext(chain)
-            return created
-        } else {
-            current.addUpstream(up)
-        }
-        return current
-    }
-
-    @Scheduled(fixedRate = 15000)
-    fun printStatuses() {
-        chainMapping.forEach { it.value.printStatus() }
-    }
-
-    override fun getAvailable(): List<Chain> {
-        return Collections.unmodifiableList(chainMapping.keys.toList())
-    }
-
-    override fun observeChains(): Flux<Chain> {
-        return Flux.merge(
-                Flux.fromIterable(getAvailable()),
-                Flux.from(chainsBus)
         )
+        log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
+        ds.start()
+                .doOnNext {
+                    log.info("Chain ${it.chain} has ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}")
+                }
+                .subscribe(currentUpstreams::update)
     }
 
-    override fun getDefaultMethods(chain: Chain): CallMethods {
-        var current = callTargets[chain]
-        if (current == null) {
-            current = QuorumBasedMethods(objectMapper, chain)
-            callTargets[chain] = current
-        }
-        return current
-    }
 
-    override fun isAvailable(chain: Chain): Boolean {
-        return chainMapping.containsKey(chain) && callTargets.containsKey(chain)
-    }
 }

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

@@ -0,0 +1,98 @@
+/**
+ * Copyright (c) 2019 ETCDEV GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.grpc.Chain
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Repository
+import reactor.core.publisher.Flux
+import reactor.core.publisher.TopicProcessor
+import java.util.*
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+@Repository
+class CurrentUpstreams(
+        @Autowired private val objectMapper: ObjectMapper
+): Upstreams {
+
+    private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java)
+
+    private val chainMapping = ConcurrentHashMap<Chain, ChainUpstreams>()
+    private val chainsBus = TopicProcessor.create<Chain>()
+    private val callTargets = HashMap<Chain, QuorumBasedMethods>()
+    private val updateLock = ReentrantLock()
+
+    fun update(change: UpstreamChange) {
+        updateLock.withLock {
+            val chain = change.chain
+            val up = change.upstream
+            val current = chainMapping[chain]
+            if (change.type == UpstreamChange.ChangeType.REMOVED) {
+                current?.removeUpstream(up.getId())
+                log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed")
+            } else {
+                if (current == null) {
+                    val created = ChainUpstreams(chain, ArrayList<Upstream>(), objectMapper)
+                    created.addUpstream(up)
+                    created.start()
+                    chainMapping[chain] = created
+                    chainsBus.onNext(chain)
+                } else {
+                    current.addUpstream(up)
+                }
+                log.info("Upstream ${change.upstream.getId()} with chain $chain has been added")
+            }
+        }
+    }
+
+    override fun getUpstream(chain: Chain): AggregatedUpstream? {
+        return chainMapping[chain]
+    }
+
+    @Scheduled(fixedRate = 15000)
+    fun printStatuses() {
+        chainMapping.forEach { it.value.printStatus() }
+    }
+
+    override fun getAvailable(): List<Chain> {
+        return Collections.unmodifiableList(chainMapping.keys.toList())
+    }
+
+    override fun observeChains(): Flux<Chain> {
+        return Flux.merge(
+                Flux.fromIterable(getAvailable()),
+                Flux.from(chainsBus)
+        )
+    }
+
+    override fun getDefaultMethods(chain: Chain): CallMethods {
+        var current = callTargets[chain]
+        if (current == null) {
+            current = QuorumBasedMethods(objectMapper, chain)
+            callTargets[chain] = current
+        }
+        return current
+    }
+
+    override fun isAvailable(chain: Chain): Boolean {
+        return chainMapping.containsKey(chain) && callTargets.containsKey(chain)
+    }
+}

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

@@ -31,4 +31,5 @@ interface Upstream {
     fun getLag(): Long
     fun getLabels(): Collection<UpstreamsConfig.Labels>
     fun getMethods(): CallMethods
+    fun getId(): String
 }

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

@@ -0,0 +1,31 @@
+/**
+ * Copyright (c) 2019 ETCDEV GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.grpc.Chain
+
+class UpstreamChange(
+        val chain: Chain,
+        val upstream: Upstream,
+        val type: ChangeType
+) {
+    enum class ChangeType {
+        ADDED,
+        REVALIDATED,
+        STALE,
+        REMOVED,
+    }
+}

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

@@ -19,7 +19,6 @@ import io.emeraldpay.grpc.Chain
 import reactor.core.publisher.Flux
 
 interface Upstreams {
-    fun addUpstream(chain: Chain, up: Upstream): AggregatedUpstream
     fun getUpstream(chain: Chain): AggregatedUpstream?
     fun getAvailable(): List<Chain>
     fun observeChains(): Flux<Chain>

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

@@ -23,6 +23,7 @@ import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 
 open class EthereumUpstream(
+        private val id: String,
         val chain: Chain,
         private val api: DirectEthereumApi,
         private val ethereumWs: EthereumWs? = null,
@@ -31,7 +32,7 @@ open class EthereumUpstream(
         private val targets: CallMethods
 ): DefaultUpstream(), Lifecycle {
 
-    constructor(chain: Chain, api: DirectEthereumApi): this(chain, api, null,
+    constructor(id: String, chain: Chain, api: DirectEthereumApi): this(id, chain, api, null,
             UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()),
             DirectCallMethods())
 
@@ -45,6 +46,10 @@ open class EthereumUpstream(
         api.upstream = this
     }
 
+    override fun getId(): String {
+        return id
+    }
+
     override fun start() {
         log.info("Configured for ${chain.chainName}")
 

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

@@ -45,6 +45,7 @@ import java.util.function.Function
 import kotlin.collections.ArrayList
 
 open class GrpcUpstream(
+        private val parentId: String,
         private val chain: Chain,
         private val client: ReactorBlockchainGrpc.ReactorBlockchainStub,
         private val objectMapper: ObjectMapper
@@ -58,12 +59,13 @@ open class GrpcUpstream(
     private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
     private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
     private val head = Head(this)
-    private var targets: CallMethods = DirectCallMethods()
+    private var targets: CallMethods? = null
     private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper)
 
     private var headSubscription: Disposable? = null
 
     open fun createApi(matcher: Selector.Matcher): DirectEthereumApi {
+        val targets = this.getMethods()
         val rpcClient = DefaultRpcClient(grpcTransport.withLabels(Selector.extractLabels(matcher)))
         return DirectEthereumApi(rpcClient, objectMapper, targets).let {
             it.upstream = this
@@ -71,7 +73,12 @@ open class GrpcUpstream(
         }
     }
 
+    override fun getId(): String {
+        return "$parentId/${chain.chainCode}"
+    }
+
     override fun start() {
+        if (this.isRunning) return
         val chainRef = Common.Chain.newBuilder()
                 .setTypeValue(chain.id)
                 .build()
@@ -116,7 +123,12 @@ open class GrpcUpstream(
                             .executeAndConvert(Commands.eth().getBlock(it.hash))
                             .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream")))
                             .doOnError { t ->
-                                log.warn("Failed to download block data", t)
+                                val msg = "Failed to download block data for chain $chain"
+                                if (t is RpcException) {
+                                    log.warn("$msg. Message: ${t.message}")
+                                } else {
+                                    log.error(msg, t)
+                                }
                             }
                 }
                 .onErrorContinue { err, _ ->
@@ -134,9 +146,9 @@ open class GrpcUpstream(
         targets = DirectCallMethods(conf.supportedMethodsList.toSet())
         val nodes = NodeDetailsList()
         val allLabels = ArrayList<UpstreamsConfig.Labels>()
-        conf.nodesList.forEach { node ->
+        conf.nodesList.forEach { remoteNode ->
-            val node = NodeDetailsList.NodeDetails(node.quorum,
+            val node = NodeDetailsList.NodeDetails(remoteNode.quorum,
-                    node.labelsList.let { provided ->
+                    remoteNode.labelsList.let { provided ->
                         val labels = UpstreamsConfig.Labels()
                         provided.forEach {
                             labels[it.name] = it.value
@@ -171,7 +183,7 @@ open class GrpcUpstream(
     }
 
     override fun getMethods(): CallMethods {
-        return targets
+        return targets ?: throw IllegalStateException("Upstream is not initialized yet")
     }
 
     override fun isAvailable(): Boolean {

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

@@ -19,7 +19,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.upstream.Upstreams
+import io.emeraldpay.dshackle.upstream.UpstreamChange
 import io.emeraldpay.grpc.Chain
 import io.grpc.ManagedChannelBuilder
 import io.grpc.netty.NettyChannelBuilder
@@ -27,15 +27,14 @@ import io.netty.handler.ssl.*
 import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
-import reactor.core.publisher.toFlux
-import reactor.util.function.Tuple2
-import reactor.util.function.Tuples
 import java.io.File
+import java.time.Duration
 import java.util.*
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
 
 class GrpcUpstreams(
+        private val id: String,
         private val host: String,
         private val port: Int,
         private val objectMapper: ObjectMapper,
@@ -44,39 +43,32 @@ class GrpcUpstreams(
     private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
 
     private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null
-    private var known = HashMap<Chain, GrpcUpstream>()
+    private val known = HashMap<Chain, GrpcUpstream>()
     private val lock = ReentrantLock()
 
-    fun start(): Flux<Tuple2<Chain, GrpcUpstream>> {
+    fun start(): Flux<UpstreamChange> {
         val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) {
             NettyChannelBuilder.forAddress(host, port)
                     .useTransportSecurity()
                     .sslContext(withTls(auth))
         } else {
-            log.warn("Using insecure connection for $host:$port")
+            log.warn("Using insecure connection to $host:$port")
             ManagedChannelBuilder.forAddress(host, port)
                     .usePlaintext()
         }
 
         val client = ReactorBlockchainGrpc.newReactorStub(channel.build())
         this.client = client
-        val loaded = client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
+
-                .map { value ->
+        val updates = Flux.interval(Duration.ZERO, Duration.ofMinutes(1))
-                    val chains = ArrayList<Chain>()
+                .flatMap {
-                    value.chainsList.filter {
+                    client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
-                        Chain.byId(it.chain.number) != Chain.UNSPECIFIED
+                }.flatMap { value ->
-                    }.map { chainDetails ->
+                    processDescription(value)
-                        val chain = Chain.byId(chainDetails.chain.number)
-                        val up = getOrCreate(chain)
-                        up.init(chainDetails)
-                        chains.add(chain)
-                        Tuples.of(chain, up)
-                    }
-                }.flatMapMany {
-                    it.toFlux()
                 }.doOnError { t ->
                     log.error("Failed to get description from $host:$port", t)
                 }
+
         //TODO subscribe only after receiving details
         client.subscribeStatus(BlockchainOuterClass.StatusRequest.newBuilder().build())
                 .subscribe { value ->
@@ -85,7 +77,30 @@ class GrpcUpstreams(
                         known[chain]?.onStatus(value)
                     }
                 }
-        return loaded
+        return updates
+    }
+
+    fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux<UpstreamChange> {
+        val current = value.chainsList.filter {
+            Chain.byId(it.chain.number) != Chain.UNSPECIFIED
+        }.map { chainDetails ->
+            val chain = Chain.byId(chainDetails.chain.number)
+            val up = getOrCreate(chain)
+            (up.upstream as GrpcUpstream).init(chainDetails)
+            up
+        }
+
+        val added = current.filter {
+            it.type == UpstreamChange.ChangeType.ADDED
+        }
+
+        val removed = known.filterNot { kv ->
+            val stillCurrent = current.any { c -> c.chain == kv.key }
+            stillCurrent
+        }.map {
+            UpstreamChange(it.key, known.remove(it.key)!!, UpstreamChange.ChangeType.REMOVED)
+        }
+        return Flux.fromIterable(removed + added)
     }
 
     internal fun withTls(auth: UpstreamsConfig.TlsAuth): SslContext {
@@ -106,16 +121,16 @@ class GrpcUpstreams(
         return sslContext.build()
     }
 
-    fun getOrCreate(chain: Chain): GrpcUpstream {
+    fun getOrCreate(chain: Chain): UpstreamChange {
         lock.withLock {
             val current = known[chain]
             return if (current == null) {
-                val created = GrpcUpstream(chain, client!!, objectMapper)
+                val created = GrpcUpstream(id, chain, client!!, objectMapper)
                 known[chain] = created
                 created.start()
-                created
+                UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED)
             } else {
-                current
+                UpstreamChange(chain, current, UpstreamChange.ChangeType.REVALIDATED)
             }
         }
     }

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -15,6 +15,10 @@
  */
 package io.emeraldpay.dshackle.config
 
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.rpc.RpcClient
 import spock.lang.Specification
 
 class UpstreamsConfigReaderSpec extends Specification {
@@ -194,4 +198,31 @@ class UpstreamsConfigReaderSpec extends Specification {
             }
         }
     }
+
+    def "Parse config with invalid ids"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("upstreams-no-id.yaml")
+        when:
+        def act = reader.read(config)
+        then:
+        act != null
+        act.upstreams.size() == 1
+        with(act.upstreams.get(0)) {
+            id == "test"
+        }
+    }
+
+    def "Invalidate wrong ids"() {
+        expect:
+        !reader.isValid(new UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>(id: id))
+        where:
+        id << ["", null, "a", "ab", "!ab", "foo bar", "foo@bar", "123test", "_test", "test/test"]
+    }
+
+    def "Accept good ids"() {
+        expect:
+        reader.isValid(new UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>(id: id))
+        where:
+        id << ["test", "test_test", "test-test", "test123", "test1test", "foo_bar_12"]
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy

@@ -36,8 +36,16 @@ class EthereumUpstreamMock extends EthereumUpstream {
         this(chain, api, new QuorumBasedMethods(TestingCommons.objectMapper(), chain))
     }
 
+    EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull DirectEthereumApi api) {
+        this(id, chain, api, new QuorumBasedMethods(TestingCommons.objectMapper(), chain))
+    }
+
     EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api, CallMethods methods) {
-        super(chain, api, null,
+        this("test", chain, api, methods)
+    }
+
+    EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull DirectEthereumApi api, CallMethods methods) {
+        super(id, chain, api, null,
                 UpstreamsConfig.Options.getDefaults(), new NodeDetailsList.NodeDetails(1, new UpstreamsConfig.Labels()),
                 methods)
         setLag(0)

src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy

@@ -38,7 +38,6 @@ class UpstreamsMock implements Upstreams {
         addUpstream(chain2, up2)
     }
 
-    @Override
     AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) {
         if (!upstreams.containsKey(chain)) {
             upstreams[chain] = new ChainUpstreams(chain, [up], TestingCommons.objectMapper())

src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy

@@ -0,0 +1,43 @@
+/**
+ * Copyright (c) 2019 ETCDEV GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.quorum.AlwaysQuorum
+import io.emeraldpay.dshackle.test.EthereumUpstreamMock
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
+import io.emeraldpay.grpc.Chain
+import spock.lang.Specification
+
+class AggregatedUpstreamSpec extends Specification {
+
+    def "Aggregates methods"() {
+        setup:
+        def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test1", "eth_test2"]))
+        def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test2", "eth_test3"]))
+        def aggr = new ChainUpstreams(Chain.ETHEREUM, [up1, up2], TestingCommons.objectMapper())
+        when:
+        aggr.onUpstreamsUpdated()
+        def act = aggr.getMethods()
+        then:
+        act.isAllowed("eth_test1")
+        act.isAllowed("eth_test2")
+        act.isAllowed("eth_test3")
+        act.getQuorumFor("eth_test1") instanceof AlwaysQuorum
+        act.getQuorumFor("eth_test2") instanceof AlwaysQuorum
+        act.getQuorumFor("eth_test3") instanceof AlwaysQuorum
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy

@@ -0,0 +1,56 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.test.EthereumUpstreamMock
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.rpc.RpcClient
+import spock.lang.Specification
+
+class CurrentUpstreamsSpec extends Specification {
+
+    def "add upstream"() {
+        setup:
+        def current = new CurrentUpstreams(TestingCommons.objectMapper())
+        def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient)))
+        when:
+        current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED))
+        then:
+        current.getAvailable() == [Chain.ETHEREUM]
+        current.getUpstream(Chain.ETHEREUM).getAll()[0] == up
+    }
+
+    def "add multiple upstreams"() {
+        setup:
+        def current = new CurrentUpstreams(TestingCommons.objectMapper())
+        def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient)))
+        def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(RpcClient)))
+        def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient)))
+        when:
+        current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED))
+        current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED))
+        current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED))
+        then:
+        current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet()
+        current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up1, up3].toSet()
+        current.getUpstream(Chain.ETHEREUM_CLASSIC).getAll().toSet() == [up2].toSet()
+    }
+
+    def "remove upstream"() {
+        setup:
+        def current = new CurrentUpstreams(TestingCommons.objectMapper())
+        def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient)))
+        def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(RpcClient)))
+        def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient)))
+        def up1_del = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient)))
+        when:
+        current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED))
+        current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED))
+        current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED))
+        current.update(new UpstreamChange(Chain.ETHEREUM, up1_del, UpstreamChange.ChangeType.REMOVED))
+        then:
+        current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet()
+        current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up3].toSet()
+        current.getUpstream(Chain.ETHEREUM_CLASSIC).getAll().toSet() == [up2].toSet()
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy

@@ -40,6 +40,7 @@ class FilteringApiIteratorSpec extends Specification {
                 [test: "baz"]
         ].collect {
             new EthereumUpstream(
+                    "test",
                     Chain.ETHEREUM,
                     new DirectEthereumApi(rpcClient, objectMapper, ethereumTargets),
                     (EthereumWs) null,

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy

@@ -70,7 +70,7 @@ class GrpcUpstreamSpec extends Specification {
                 )
             }
         })
-        def upstream = new GrpcUpstream(chain, client, objectMapper)
+        def upstream = new GrpcUpstream("test", chain, client, objectMapper)
         upstream.setLag(0)
         upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
                 .addAllSupportedMethods(["eth_getBlockByHash"])
@@ -129,7 +129,7 @@ class GrpcUpstreamSpec extends Specification {
                 finished.complete(true)
             }
         })
-        def upstream = new GrpcUpstream(chain, client, objectMapper)
+        def upstream = new GrpcUpstream("test", chain, client, objectMapper)
         upstream.setLag(0)
         upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
                 .addAllSupportedMethods(["eth_getBlockByHash"])
@@ -189,7 +189,7 @@ class GrpcUpstreamSpec extends Specification {
                 finished.complete(true)
             }
         })
-        def upstream = new GrpcUpstream(chain, client, objectMapper)
+        def upstream = new GrpcUpstream("test", chain, client, objectMapper)
         upstream.setLag(0)
         upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
                 .addAllSupportedMethods(["eth_getBlockByHash"])

src/test/resources/upstreams-no-id.yaml

@@ -0,0 +1,26 @@
+version: v1
+
+upstreams:
+  - chain: ethereum
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+  - chain: ethereum
+    id: test
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+  - chain: ethereum
+    id: test/test
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+  - chain: ethereum
+    id: !test
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"