Merge pull request #82 from p2p-org/fix_stuck_heads

Fix stuck heads

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
@@ -69,6 +72,7 @@ open class ConfiguredUpstreams(
     override fun run(args: ApplicationArguments) {
         log.debug("Starting upstreams")
         val defaultOptions = buildDefaultOptions(config)
+        val observerScheduler = Schedulers.newParallel("status-observer", 3)
         config.upstreams.forEach { up ->
             if (!up.isEnabled) {
                 log.debug("Upstream ${up.id} is disabled")
@@ -103,15 +107,22 @@ open class ConfiguredUpstreams(
                             options
                         )
                     }
-
-                    else -> {
-                        log.error("Chain is unsupported: ${up.chain}")
-                        return@forEach
-                    }
                 }
                 upstream?.let {
-                    val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED)
+                    Flux.concat(Mono.just(UpstreamChangeEvent.ChangeType.ADDED), upstream.observeStatus())
-                    eventPublisher.publishEvent(event)
+                        .distinctUntilChanged()
+                        .subscribeOn(observerScheduler)
+                        .subscribe { status ->
+                            when (status) {
+                                UpstreamAvailability.UNAVAILABLE -> UpstreamChangeEvent.ChangeType.REMOVED
+                                else -> UpstreamChangeEvent.ChangeType.REVALIDATED
+                            }.let { eventType ->
+                                if (eventType == UpstreamChangeEvent.ChangeType.REMOVED) {
+                                    log.warn("Remove upstream ${it::class.java.simpleName}:${upstream.getId()} due to $status")
+                                }
+                                eventPublisher.publishEvent(UpstreamChangeEvent(chain, upstream, eventType))
+                            }
+                        }
                 }
             }
         }

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

@@ -17,6 +17,8 @@ package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
+import io.micrometer.core.instrument.Gauge
+import io.micrometer.core.instrument.Metrics
 import org.slf4j.LoggerFactory
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
@@ -28,6 +30,7 @@ import reactor.core.scheduler.Schedulers
 import reactor.kotlin.core.publisher.toMono
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantLock
 
 abstract class AbstractHead @JvmOverloads constructor(
@@ -49,18 +52,20 @@ abstract class AbstractHead @JvmOverloads constructor(
     private val lock = ReentrantLock()
 
     init {
+        val state = AtomicBoolean(false)
+        Gauge.builder("stuck_head", state) {
+            if (it.get()) 1.0 else 0.0
+        }.tag("upstream", upstreamId).tag("class", this.javaClass.simpleName).register(Metrics.globalRegistry)
+        Gauge.builder("current_head", forkChoice) {
+            it.getHead()?.height?.toDouble() ?: 0.0
+        }.tag("upstream", upstreamId).tag("class", this.javaClass.simpleName).register(Metrics.globalRegistry)
         Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
             {
                 val delay = System.currentTimeMillis() - lastHeadUpdated
-                if (delay > awaitHeadTimeoutMs) {
+                val delayed = delay > awaitHeadTimeoutMs
-                    log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass} - restart")
+                state.set(delayed)
-                    if (lock.tryLock()) {
+                if (delayed) {
-                        try {
+                    log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass}")
-                            start()
-                        } finally {
-                            lock.unlock()
-                        }
-                    }
                 }
             }, 300, 30, TimeUnit.SECONDS
         )

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

@@ -39,7 +39,7 @@ import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import java.time.Duration
 import java.time.Instant
-import java.util.*
+import java.util.Locale
 import java.util.concurrent.atomic.AtomicReference
 import java.util.concurrent.locks.ReentrantLock
 import java.util.function.Predicate
@@ -63,6 +63,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")
@@ -71,6 +72,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 ->
@@ -118,19 +120,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.getId() == id).also {
+                if (it) {
+                    removed[id] = up
+                }
+            }
+        }.also {
+            if (it) {
+                onUpstreamsUpdated()
+                setHead(updateHead())
+            }
         }
-    }
 
     /**
      * Get a source for direct APIs
@@ -283,27 +298,26 @@ 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() }
 
-        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], weak=[$weak]")
+        val instance = System.identityHashCode(this).toString(16)
+        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], weak=[$weak] ($instance)")
     }
 
     fun test(event: UpstreamChangeEvent): Boolean {
@@ -315,18 +329,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.warn("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.info("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()
             }
     }