diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 11e07ec1..74237112 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/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() private var completed = false private val beforeBlockHandlers = ArrayList() + private val lastUpdateTime = AtomicLong(0L) fun follow(source: Flux): 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() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt index 75d66b27..cdc46385 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt +++ b/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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt index ec366dff..97696c0e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index 5bcce942..4181e1d3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/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 = 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? = + override fun tryProxy( + matcher: Selector.Matcher, + request: BlockchainOuterClass.NativeSubscribeRequest + ): Flux? = upstreams.filter { matcher.matches(it) }.takeIf { ups -> diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy index d5d1d659..4972bd8e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy +++ b/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 + } }