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