added watchdog for Eth head

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

@@ -23,6 +23,7 @@ import reactor.core.publisher.Flux
 import reactor.core.publisher.Sinks
 import reactor.core.scheduler.Schedulers
 import reactor.kotlin.core.publisher.toMono
+import java.util.concurrent.atomic.AtomicLong
 
 abstract class AbstractHead(
     private val forkChoice: ForkChoice,
@@ -36,6 +37,7 @@ abstract class AbstractHead(
     private var stream = Sinks.many().multicast().directBestEffort<BlockContainer>()
     private var completed = false
     private val beforeBlockHandlers = ArrayList<Runnable>()
+    private val lastUpdateTime = AtomicLong(0L)
 
     fun follow(source: Flux<BlockContainer>): Disposable {
         if (completed) {
@@ -72,6 +74,7 @@ abstract class AbstractHead(
                             if (result.isFailure && result != Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) {
                                 log.warn("Failed to dispatch block: $result as ${this.javaClass}")
                             }
+                            lastUpdateTime.set(System.currentTimeMillis())
                         }
 
                         is ForkChoice.ChoiceResult.Same -> {}
@@ -111,4 +114,7 @@ abstract class AbstractHead(
     override fun getCurrentHeight(): Long? {
         return getCurrent()?.height
     }
+
+    override fun getLastUpdateTime(): Long =
+        lastUpdateTime.get()
 }

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

@@ -31,4 +31,6 @@ class EmptyHead : Head {
     override fun getCurrentHeight(): Long? {
         return null
     }
+
+    override fun getLastUpdateTime(): Long = 0L
 }

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

@@ -37,4 +37,5 @@ interface Head {
     fun onBeforeBlock(handler: Runnable)
 
     fun getCurrentHeight(): Long?
+    fun getLastUpdateTime(): Long
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -37,6 +37,9 @@ import org.springframework.context.Lifecycle
 import org.springframework.util.ConcurrentReferenceHashMap
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.locks.ReentrantLock
 
 @Suppress("UNCHECKED_CAST")
 open class EthereumPosMultiStream(
@@ -56,6 +59,7 @@ open class EthereumPosMultiStream(
     private val subscribe = EthereumSubscribe(this)
     private val filteredHeads: MutableMap<String, Head> =
         ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
+    private val lock = ReentrantLock()
 
     init {
         this.init()
@@ -66,6 +70,20 @@ open class EthereumPosMultiStream(
             head = updateHead()
         }
         super.init()
+        Executors.newScheduledThreadPool(1).scheduleAtFixedRate({
+            val timeout = System.currentTimeMillis() - (head?.getLastUpdateTime() ?: 0L)
+            log.debug("Check head is active! Lst updated $timeout ms ago")
+            if (timeout > 60_000 && lock.tryLock()) {
+                log.warn("Timeout is over 1 min - restart head")
+                try {
+                    head = updateHead()
+                } catch (e: Exception) {
+                    log.error(e.message, e)
+                } finally {
+                    lock.unlock()
+                }
+            }
+        }, 60, 30, TimeUnit.SECONDS)
     }
 
     override fun start() {
@@ -91,7 +109,10 @@ open class EthereumPosMultiStream(
         return head!!
     }
 
-    override fun tryProxy(matcher: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>? =
+    override fun tryProxy(
+        matcher: Selector.Matcher,
+        request: BlockchainOuterClass.NativeSubscribeRequest
+    ): Flux<out Any>? =
         upstreams.filter {
             matcher.matches(it)
         }.takeIf { ups ->

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

@@ -67,4 +67,9 @@ class EthereumHeadMock implements Head {
     Long getCurrentHeight() {
         return latest?.height
     }
+
+    @Override
+    long getLastUpdateTime() {
+        return 0
+    }
 }