From 8c0d3faadd6ecf0e5d9bbd87e99bb82800f8877a Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Thu, 6 Oct 2022 10:57:24 +0300 Subject: [PATCH 1/6] fixed head subscription recovery --- .../kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt | 2 ++ .../io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) 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 2b9ece14..4372e1f1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -104,6 +104,8 @@ class GrpcHead( blocks = blocks.onErrorContinue { err, _ -> log.error("Head subscription error. ${err.javaClass.name}:${err.message}", err) + }.doOnNext{ + log.info("Received block ${it.height}") } headSubscription = super.follow(blocks) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 0fafb099..446dd487 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -44,7 +44,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux -import java.net.ConnectException +import java.io.IOException import java.time.Duration import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock @@ -92,7 +92,7 @@ class GrpcUpstreams( .flatMap { client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) }.onErrorContinue { t, _ -> - if (ExceptionUtils.indexOfType(t, ConnectException::class.java) >= 0) { + if (ExceptionUtils.indexOfType(t, IOException::class.java) >= 0) { log.warn("gRPC upstream $host:$port is unavailable. (${t.javaClass}: ${t.message})") known.values.forEach { it.setStatus(UpstreamAvailability.UNAVAILABLE) From cf55c1e861aeb065b54c497ea17c670511c89f04 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Thu, 6 Oct 2022 16:33:28 +0300 Subject: [PATCH 2/6] fix lint --- .../kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 4372e1f1..8e8cba03 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -104,7 +104,7 @@ class GrpcHead( blocks = blocks.onErrorContinue { err, _ -> log.error("Head subscription error. ${err.javaClass.name}:${err.message}", err) - }.doOnNext{ + }.doOnNext { log.info("Received block ${it.height}") } From 0aacc098626010f8969db2a0a5c9bdb6865454cf Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Fri, 7 Oct 2022 18:14:42 +0300 Subject: [PATCH 3/6] added watchdog for Eth head --- .../dshackle/upstream/AbstractHead.kt | 6 +++++ .../emeraldpay/dshackle/upstream/EmptyHead.kt | 2 ++ .../io/emeraldpay/dshackle/upstream/Head.kt | 1 + .../ethereum_pos/EthereumPosMultiStream.kt | 23 ++++++++++++++++++- .../dshackle/test/EthereumHeadMock.groovy | 5 ++++ 5 files changed, 36 insertions(+), 1 deletion(-) 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 + } } From c53913db6a3adf2e0273cfcc38f131a7114a66a3 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Mon, 10 Oct 2022 13:48:34 +0300 Subject: [PATCH 4/6] 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 From d30e4236d1c96d8084a0387e8ffe27cb7aef3da9 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Mon, 10 Oct 2022 14:40:14 +0300 Subject: [PATCH 5/6] remove redundant code --- .../dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt | 2 -- 1 file changed, 2 deletions(-) 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 007d9a28..959b3485 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,7 +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.locks.ReentrantLock @Suppress("UNCHECKED_CAST") open class EthereumPosMultiStream( @@ -57,7 +56,6 @@ open class EthereumPosMultiStream( private val subscribe = EthereumSubscribe(this) private val filteredHeads: MutableMap = ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK) - private val lock = ReentrantLock() init { this.init() From 26b8a402d5bc0bfb1107434785ffdd438771a81c Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 11 Oct 2022 02:16:07 +0300 Subject: [PATCH 6/6] try fix tests --- .../dshackle/upstream/AbstractHead.kt | 20 +++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 28d8fcf2..f93152c0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux +import reactor.core.publisher.SignalType import reactor.core.publisher.Sinks import reactor.core.publisher.Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER import reactor.core.publisher.Sinks.EmitResult.OK @@ -27,6 +28,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.locks.ReentrantLock abstract class AbstractHead( private val forkChoice: ForkChoice, @@ -43,6 +45,7 @@ abstract class AbstractHead( private val beforeBlockHandlers = ArrayList() private var stopping = false private var lastHeadUpdated = 0L + private val lock = ReentrantLock() init { Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( @@ -50,7 +53,12 @@ abstract class AbstractHead( val delay = System.currentTimeMillis() - lastHeadUpdated if (delay > awaitHeadTimeoutMs) { log.warn("No head updates for $delay ms @ ${this.javaClass} - restart") - start() + try { + lock.tryLock() + start() + } finally { + lock.unlock() + } } }, 300, 30, TimeUnit.SECONDS ) @@ -71,13 +79,13 @@ 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 - if (stopping) { - log.info("Received signal $it - stop emit new head!!!") + if (it == SignalType.ON_ERROR && !stopping) { + log.warn("Received signal $it unexpectedly - restart head") + lastHeadUpdated = 0L + } else { + log.warn("Received signal $it - stop emit new head!!!") completed = true stream.tryEmitComplete() - } else { - log.warn("Received signal $it unexpectedly - restart head") - start() } } .subscribeOn(Schedulers.boundedElastic())