From c53913db6a3adf2e0273cfcc38f131a7114a66a3 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Mon, 10 Oct 2022 13:48:34 +0300 Subject: [PATCH] better stuck head handling --- .../dshackle/upstream/AbstractHead.kt | 53 ++++++++++++++----- .../emeraldpay/dshackle/upstream/EmptyHead.kt | 5 +- .../io/emeraldpay/dshackle/upstream/Head.kt | 5 +- .../dshackle/upstream/MergedHead.kt | 11 +++- .../upstream/bitcoin/BitcoinRpcHead.kt | 4 +- .../upstream/bitcoin/BitcoinZMQHead.kt | 4 +- .../upstream/ethereum/EthereumRpcHead.kt | 2 + .../upstream/ethereum/EthereumWsHead.kt | 2 + .../ethereum_pos/EthereumPosMultiStream.kt | 16 ------ .../dshackle/upstream/grpc/GrpcHead.kt | 4 +- .../dshackle/test/EthereumHeadMock.groovy | 9 +++- .../dshackle/upstream/AbstractHeadSpec.groovy | 13 +++-- .../dshackle/upstream/MergedHeadSpec.groovy | 4 +- 13 files changed, 90 insertions(+), 42 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 74237112..28d8fcf2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -21,13 +21,17 @@ import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Sinks +import reactor.core.publisher.Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER +import reactor.core.publisher.Sinks.EmitResult.OK import reactor.core.scheduler.Schedulers import reactor.kotlin.core.publisher.toMono -import java.util.concurrent.atomic.AtomicLong +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit abstract class AbstractHead( private val forkChoice: ForkChoice, - private val blockValidator: BlockValidator = BlockValidator.ALWAYS_VALID + private val blockValidator: BlockValidator = BlockValidator.ALWAYS_VALID, + awaitHeadTimeoutMs: Long = 60_000 ) : Head { companion object { @@ -37,7 +41,20 @@ abstract class AbstractHead( private var stream = Sinks.many().multicast().directBestEffort() private var completed = false private val beforeBlockHandlers = ArrayList() - private val lastUpdateTime = AtomicLong(0L) + private var stopping = false + private var lastHeadUpdated = 0L + + init { + Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( + { + val delay = System.currentTimeMillis() - lastHeadUpdated + if (delay > awaitHeadTimeoutMs) { + log.warn("No head updates for $delay ms @ ${this.javaClass} - restart") + start() + } + }, 300, 30, TimeUnit.SECONDS + ) + } fun follow(source: Flux): Disposable { if (completed) { @@ -54,8 +71,14 @@ abstract class AbstractHead( // close internal stream if upstream is finished, otherwise it gets stuck, // but technically it should never happen during normal work, only when the Head // is stopping - completed = true - stream.tryEmitComplete() + if (stopping) { + log.info("Received signal $it - stop emit new head!!!") + completed = true + stream.tryEmitComplete() + } else { + log.warn("Received signal $it unexpectedly - restart head") + start() + } } .subscribeOn(Schedulers.boundedElastic()) .subscribe { block -> @@ -69,12 +92,12 @@ abstract class AbstractHead( when (val choiceResult = forkChoice.choose(block)) { is ForkChoice.ChoiceResult.Updated -> { val newHead = choiceResult.nwhead - log.debug("New block ${newHead.height} ${newHead.hash}") - val result = stream.tryEmitNext(newHead) - if (result.isFailure && result != Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) { - log.warn("Failed to dispatch block: $result as ${this.javaClass}") + lastHeadUpdated = System.currentTimeMillis() + when (val result = stream.tryEmitNext(newHead)) { + OK -> log.debug("New block ${newHead.height} ${newHead.hash} @ ${this.javaClass}") + FAIL_ZERO_SUBSCRIBER -> log.debug("No subscribers for ${this.javaClass}") + else -> log.warn("Failed to dispatch block: $result as ${this.javaClass}") } - lastUpdateTime.set(System.currentTimeMillis()) } is ForkChoice.ChoiceResult.Same -> {} @@ -100,7 +123,6 @@ abstract class AbstractHead( } override fun getFlux(): Flux { - val curHead = forkChoice.getHead() return Flux.concat( forkChoice.getHead().toMono(), stream.asFlux() @@ -115,6 +137,11 @@ abstract class AbstractHead( return getCurrent()?.height } - override fun getLastUpdateTime(): Long = - lastUpdateTime.get() + override fun stop() { + stopping = true + } + + override fun start() { + stopping = false + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt index cdc46385..371fc95f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt @@ -31,6 +31,9 @@ class EmptyHead : Head { override fun getCurrentHeight(): Long? { return null } + override fun start() { + } - override fun getLastUpdateTime(): Long = 0L + override fun stop() { + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt index 97696c0e..3337e44a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt @@ -37,5 +37,8 @@ interface Head { fun onBeforeBlock(handler: Runnable) fun getCurrentHeight(): Long? - fun getLastUpdateTime(): Long + + fun start() + + fun stop() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt index 1518f579..42e654e0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt @@ -20,6 +20,7 @@ import com.google.common.annotations.VisibleForTesting import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux @@ -29,6 +30,10 @@ class MergedHead( forkChoice: ForkChoice ) : AbstractHead(forkChoice), Lifecycle, CachesEnabled { + companion object { + private val log = LoggerFactory.getLogger(MergedHead::class.java) + } + private var subscription: Disposable? = null override fun isRunning(): Boolean { @@ -36,16 +41,20 @@ class MergedHead( } override fun start() { + super.start() sources.forEach { head -> if (head is Lifecycle && !head.isRunning) { head.start() } } subscription?.dispose() - subscription = super.follow(Flux.merge(sources.map { it.getFlux() })) + subscription = super.follow( + Flux.merge(sources.map { it.getFlux() }).doOnNext { log.debug("New MERGED head $it") } + ) } override fun stop() { + super.stop() sources.forEach { head -> if (head is Lifecycle && head.isRunning) { head.stop() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt index 982d4fb2..72f500bc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt @@ -36,7 +36,7 @@ class BitcoinRpcHead( private val api: Reader, private val extractBlock: ExtractBlock, private val interval: Duration = Duration.ofSeconds(15) -) : Head, AbstractHead(MostWorkForkChoice()), Lifecycle { +) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinRpcHead::class.java) @@ -51,6 +51,7 @@ class BitcoinRpcHead( } override fun start() { + super.start() if (refreshSubscription != null) { log.warn("Called to start when running") return @@ -76,6 +77,7 @@ class BitcoinRpcHead( } override fun stop() { + super.stop() val copy = refreshSubscription refreshSubscription = null copy?.dispose() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt index ca916408..4ce4bcc1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt @@ -21,7 +21,7 @@ class BitcoinZMQHead( private val server: ZMQServer, private val api: Reader, private val extractBlock: ExtractBlock, -) : Head, AbstractHead(MostWorkForkChoice()), Lifecycle { +) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinZMQHead::class.java) @@ -51,11 +51,13 @@ class BitcoinZMQHead( } override fun start() { + super.start() server.start() refreshSubscription = super.follow(connect()) } override fun stop() { + super.stop() server.stop() val copy = refreshSubscription refreshSubscription = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index d299354f..00637d54 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -48,6 +48,7 @@ class EthereumRpcHead( private var refreshSubscription: Disposable? = null override fun start() { + super.start() refreshSubscription?.dispose() val base = Flux.interval(interval) .publishOn(scheduler) @@ -62,6 +63,7 @@ class EthereumRpcHead( } override fun stop() { + super.stop() refreshSubscription?.dispose() refreshSubscription = null } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index e34ef409..1532d5d8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -40,6 +40,7 @@ class EthereumWsHead( } override fun start() { + super.start() this.subscription?.dispose() val heads = Flux.merge( // get the current block, not just wait for the next update @@ -50,6 +51,7 @@ class EthereumWsHead( } override fun stop() { + super.stop() subscription?.dispose() subscription = null } 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 4181e1d3..007d9a28 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,8 +37,6 @@ 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") @@ -70,20 +68,6 @@ 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() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index 8e8cba03..ac2db057 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -87,7 +87,7 @@ class GrpcHead( log.warn("Disconnected $chain from ${parent.getId()}: ${err.message}") parent.setStatus(UpstreamAvailability.UNAVAILABLE) Mono.empty() - } + }.doFinally { log.warn("Head subscription finished: $it") } } /** @@ -116,10 +116,12 @@ class GrpcHead( } override fun start() { + super.start() this.internalStart(remote) } override fun stop() { + super.stop() headSubscription?.dispose() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy index 4972bd8e..142002b4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy @@ -69,7 +69,12 @@ class EthereumHeadMock implements Head { } @Override - long getLastUpdateTime() { - return 0 + void start() { + + } + + @Override + void stop() { + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy index 10253819..f16acc86 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy @@ -60,6 +60,7 @@ class AbstractHeadSpec extends Specification { .expectNext(blocks[1]) .then { assert called + head.stop() source.tryEmitComplete() } .expectComplete() @@ -83,7 +84,10 @@ class AbstractHeadSpec extends Specification { .expectNext(blocks[2]) .then { source.tryEmitNext(blocks[3]) } .expectNext(blocks[3]) - .then { source.tryEmitComplete() } + .then { + head.stop() + source.tryEmitComplete() + } .expectComplete() .verify(Duration.ofSeconds(1)) } @@ -110,7 +114,10 @@ class AbstractHeadSpec extends Specification { .then { source.tryEmitNext(wrongblock) } .then { source.tryEmitNext(blocks[3]) } .expectNext(blocks[3]) - .then { source.tryEmitComplete() } + .then { + head.stop() + source.tryEmitComplete() + } .expectComplete() .verify(Duration.ofSeconds(1)) } @@ -132,7 +139,7 @@ class AbstractHeadSpec extends Specification { BlockContainer getHead() { return null } - }, new BlockValidator.AlwaysValid()) + }, new BlockValidator.AlwaysValid(), 100_000) } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy index a5b747ef..e1b72360 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy @@ -46,14 +46,14 @@ class MergedHeadSpec extends Specification { class TestHead1 extends AbstractHead { TestHead1() { - super(new MostWorkForkChoice()) + super(new MostWorkForkChoice(), new BlockValidator.AlwaysValid(), 100_000) } } class TestHead2 extends AbstractHead implements Lifecycle { TestHead2() { - super(new MostWorkForkChoice()) + super(new MostWorkForkChoice(), new BlockValidator.AlwaysValid(), 100_000) } @Override