diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 11e07ec1..f93152c0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -20,13 +20,20 @@ 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 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, - private val blockValidator: BlockValidator = BlockValidator.ALWAYS_VALID + private val blockValidator: BlockValidator = BlockValidator.ALWAYS_VALID, + awaitHeadTimeoutMs: Long = 60_000 ) : Head { companion object { @@ -36,6 +43,26 @@ abstract class AbstractHead( private var stream = Sinks.many().multicast().directBestEffort() private var completed = false private val beforeBlockHandlers = ArrayList() + private var stopping = false + private var lastHeadUpdated = 0L + private val lock = ReentrantLock() + + init { + Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( + { + val delay = System.currentTimeMillis() - lastHeadUpdated + if (delay > awaitHeadTimeoutMs) { + log.warn("No head updates for $delay ms @ ${this.javaClass} - restart") + try { + lock.tryLock() + start() + } finally { + lock.unlock() + } + } + }, 300, 30, TimeUnit.SECONDS + ) + } fun follow(source: Flux): Disposable { if (completed) { @@ -52,8 +79,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 (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() + } } .subscribeOn(Schedulers.boundedElastic()) .subscribe { block -> @@ -67,10 +100,11 @@ 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}") } } @@ -97,7 +131,6 @@ abstract class AbstractHead( } override fun getFlux(): Flux { - val curHead = forkChoice.getHead() return Flux.concat( forkChoice.getHead().toMono(), stream.asFlux() @@ -111,4 +144,12 @@ abstract class AbstractHead( override fun getCurrentHeight(): Long? { return getCurrent()?.height } + + 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 75d66b27..371fc95f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt @@ -31,4 +31,9 @@ class EmptyHead : Head { override fun getCurrentHeight(): Long? { return null } + override fun start() { + } + + 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 ec366dff..3337e44a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt @@ -37,4 +37,8 @@ interface Head { fun onBeforeBlock(handler: Runnable) fun getCurrentHeight(): 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 5bcce942..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 @@ -91,7 +91,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/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index 2b9ece14..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") } } /** @@ -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) @@ -114,10 +116,12 @@ class GrpcHead( } override fun start() { + super.start() this.internalStart(remote) } override fun stop() { + super.stop() headSubscription?.dispose() } } 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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy index d5d1d659..142002b4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy @@ -67,4 +67,14 @@ class EthereumHeadMock implements Head { Long getCurrentHeight() { return latest?.height } + + @Override + 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