remove unavailable upstream from multistream`

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

@@ -48,6 +48,9 @@ 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.publisher.Mono
+import reactor.core.scheduler.Schedulers
 import java.net.URI
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.function.Function
@@ -110,8 +113,20 @@ open class ConfiguredUpstreams(
                     }
                 }
                 upstream?.let {
-                    val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED)
+                    Flux.concat(Mono.just(UpstreamChangeEvent.ChangeType.ADDED), upstream.observeStatus())
-                    eventPublisher.publishEvent(event)
+                        .distinctUntilChanged()
+                        .subscribeOn(Schedulers.boundedElastic())
+                        .subscribe { status ->
+                            when (status) {
+                                UpstreamAvailability.UNAVAILABLE -> UpstreamChangeEvent.ChangeType.REMOVED
+                                else -> UpstreamChangeEvent.ChangeType.REVALIDATED
+                            }.let { eventType ->
+                                if (eventType == UpstreamChangeEvent.ChangeType.REMOVED) {
+                                    log.warn("Remove upstream ${upstream.getId()} due to $it")
+                                }
+                                eventPublisher.publishEvent(UpstreamChangeEvent(chain, upstream, eventType))
+                            }
+                        }
                 }
             }
         }

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

@@ -43,6 +43,9 @@ import java.util.*
 import java.util.concurrent.atomic.AtomicReference
 import java.util.concurrent.locks.ReentrantLock
 import java.util.function.Predicate
+import kotlin.collections.ArrayList
+import kotlin.collections.HashMap
+import kotlin.collections.HashSet
 import kotlin.concurrent.withLock
 
 /**
@@ -64,6 +67,7 @@ abstract class Multistream(
 
     private var cacheSubscription: Disposable? = null
     private val reconfigLock = ReentrantLock()
+    private val eventLock = ReentrantLock()
     private var callMethods: CallMethods? = null
     private var callMethodsFactory: Factory<CallMethods> = Factory {
         return@Factory callMethods ?: throw FunctorException("Not initialized yet")
@@ -72,6 +76,7 @@ abstract class Multistream(
     protected var lagObserver: HeadLagObserver? = null
     private var subscription: Disposable? = null
     private var capabilities: Set<Capability> = emptySet()
+    private val removed: MutableMap<String, Upstream> = HashMap()
 
     init {
         UpstreamAvailability.values().forEach { status ->
@@ -119,19 +124,32 @@ abstract class Multistream(
     /**
      * Add an upstream
      */
-    fun addUpstream(upstream: Upstream) {
+    fun addUpstream(upstream: Upstream): Boolean =
-        upstreams.add(upstream)
+        upstreams.none {
-        onUpstreamsUpdated()
+            it.getId() == upstream.getId()
-        setHead(updateHead())
+        }.also {
-        monitorUpstream(upstream)
+            if (it) {
-    }
+                upstreams.add(upstream)
-
+                removed.remove(upstream.getId())
-    fun removeUpstream(id: String) {
+                onUpstreamsUpdated()
-        if (upstreams.removeIf { it.getId() == id }) {
+                setHead(updateHead())
-            onUpstreamsUpdated()
+                monitorUpstream(upstream)
-            setHead(updateHead())
+            }
         }
-    }
+
+    fun removeUpstream(id: String): Boolean =
+        upstreams.removeIf { up ->
+            up.takeIf { up.getId() == id }
+                ?.also { removed[id] = up }
+                ?.let { true }
+                ?: false
+        }.also {
+            if (it) {
+                onUpstreamsUpdated()
+                setHead(updateHead())
+            }
+        }
+
 
     /**
      * Get a source for direct APIs
@@ -301,23 +319,21 @@ abstract class Multistream(
         } catch (e: Exception) {
             log.warn("Head processing error: ${e.javaClass} ${e.message}")
         }
-        val statuses = upstreams.map { it.getStatus() }
+        val statuses = upstreams.asSequence().plus(removed.values).map { it.getStatus() }
             .groupBy { it }
             .map { "${it.key.name}/${it.value.size}" }
             .joinToString(",")
-        val lag = upstreams
+        val lag = upstreams.plus(removed.values).joinToString(", ") {
-            .map {
+            // by default, when no lag is available it uses Long.MAX_VALUE, and it doesn't make sense to print
-                // by default, when no lag is available it uses Long.MAX_VALUE, and it doesn't make sense to print
+            // status with such value. use NA (as Not Available) instead
-                // status with such value. use NA (as Not Available) instead
+            val value = it.getLag()
-                val value = it.getLag()
+            if (value == Long.MAX_VALUE) {
-                if (value == Long.MAX_VALUE) {
+                "NA"
-                    "NA"
+            } else {
-                } else {
+                value.toString()
-                    value.toString()
-                }
             }
-            .joinToString(", ")
+        }
-        val weak = upstreams
+        val weak = upstreams.plus(removed.values)
             .filter { it.getStatus() != UpstreamAvailability.OK }
             .joinToString(", ") { it.getId() }
 
@@ -333,18 +349,22 @@ abstract class Multistream(
     fun onUpstreamChange(event: UpstreamChangeEvent) {
         val chain = event.chain
         if (this.chain == chain) {
-            if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) {
+            eventLock.withLock {
-                removeUpstream(event.upstream.getId())
+                if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) {
-                log.error("Upstream ${event.upstream.getId()} with chain $chain has been removed")
+                    removeUpstream(event.upstream.getId()).takeIf { it }?.let {
-            } else {
+                        log.error("Upstream ${event.upstream.getId()} with chain $chain has been removed")
-                if (event.upstream is CachesEnabled) {
+                    }
-                    event.upstream.setCaches(caches)
+                } else {
+                    if (event.upstream is CachesEnabled) {
+                        event.upstream.setCaches(caches)
+                    }
+                    addUpstream(event.upstream).takeIf { it }?.let {
+                        if (!started) {
+                            start()
+                        }
+                        log.error("Upstream ${event.upstream.getId()} with chain $chain has been added")
+                    }
                 }
-                addUpstream(event.upstream)
-                if (!started) {
-                    start()
-                }
-                log.error("Upstream ${event.upstream.getId()} with chain $chain has been added")
             }
         }
     }

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

@@ -61,7 +61,7 @@ open class DefaultEthereumHead(
                 BlockContainer.fromEthereumJson(it.getResult(), upstreamId)
             }
             .onErrorResume { err ->
-                log.debug("Failed to fetch latest block: ${err.message}")
+                log.error("Failed to fetch latest block: ${err.message}")
                 Mono.empty()
             }
     }