StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #26 from p2p-org/fix_grpc_head_subscription_recovery

Browse Source 
fixed head subscription recovery
This commit is contained in:
Vyacheslav Shebanov
2022-10-11 13:19:56 +04:00
committed by GitHub
parent 47a36f7831 26b8a402d5
commit 07d97029a5
14 changed files with 111 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt
View File

@@ -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<BlockContainer>()
private var completed = false
private val beforeBlockHandlers = ArrayList<Runnable>()
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<BlockContainer>): 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<BlockContainer> {
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
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt
View File

@@ -31,4 +31,9 @@ class EmptyHead : Head {
override fun getCurrentHeight(): Long? {
return null
}
override fun start() {
}
override fun stop() {
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt
View File

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

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt
View File

@@ -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()

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
View File

@@ -36,7 +36,7 @@ class BitcoinRpcHead(
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
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()

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
View File

@@ -21,7 +21,7 @@ class BitcoinZMQHead(
private val server: ZMQServer,
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
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

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
View File

@@ -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
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -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
}

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

@@ -91,7 +91,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 ->

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
View File

@@ -87,7 +87,7 @@ class GrpcHead(
log.warn("Disconnected $chain from ${parent.getId()}: ${err.message}")
parent.setStatus(UpstreamAvailability.UNAVAILABLE)
Mono.empty<BlockchainOuterClass.ChainHead>()
}
}.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()
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -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)

10
src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy
View File

@@ -67,4 +67,14 @@ class EthereumHeadMock implements Head {
Long getCurrentHeight() {
return latest?.height
}
@Override
void start() {
}
@Override
void stop() {
}
}

13
src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy
View File

@@ -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)
}
}
}

4
src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy
View File

@@ -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