use head scheduler instead of stock scheduler (#206)

2023-04-21 15:53:21 +08:00
parent 39b110c3c0
commit f218e6e73b
38 changed files with 251 additions and 139 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt
@@ -21,18 +21,17 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
    open fun allMultistreams(
        cachesFactory: CachesFactory,
        callTargetsHolder: CallTargetsHolder,
-        @Qualifier("headMergedScheduler")
+        @Qualifier("headScheduler")
        headScheduler: Scheduler,
        tracer: Tracer
    ): List<Multistream> {
        return Chain.values()
            .filterNot { it == Chain.UNSPECIFIED }
-            .mapNotNull { chain ->
+            .map { chain ->
                when (BlockchainType.from(chain)) {
                    BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory, headScheduler, tracer)
                    BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory, headScheduler, tracer)
-                    BlockchainType.BITCOIN -> bitcoinMultistream(chain, cachesFactory)
-                    else -> null
+                    BlockchainType.BITCOIN -> bitcoinMultistream(chain, cachesFactory, headScheduler)
                }
            }
    }
@@ -73,14 +72,16 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)

    open fun bitcoinMultistream(
        chain: Chain,
-        cachesFactory: CachesFactory
+        cachesFactory: CachesFactory,
+        headScheduler: Scheduler
    ): BitcoinMultistream {
        val name = "multi-bitcoin-$chain"

        return BitcoinMultistream(
            chain,
            ArrayList(),
-            cachesFactory.getCaches(chain)
+            cachesFactory.getCaches(chain),
+            headScheduler
        ).also { register(it, name) }
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt
@@ -26,8 +26,8 @@ open class SchedulersConfig {
    }

    @Bean
-    open fun headMergedScheduler(monitoringConfig: MonitoringConfig): Scheduler {
-        return makeScheduler("head-scheduler", 5, monitoringConfig)
+    open fun headScheduler(monitoringConfig: MonitoringConfig): Scheduler {
+        return makeScheduler("head-scheduler", 4, monitoringConfig)
    }

    @Bean
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -82,7 +82,9 @@ open class ConfiguredUpstreams(
    private val grpcTracing: GrpcTracing,
    private val wsConnectionResubscribeScheduler: Scheduler,
    @Autowired(required = false)
-    private val clientSpansInterceptor: ClientInterceptor?
+    private val clientSpansInterceptor: ClientInterceptor?,
+    @Qualifier("headScheduler")
+    private val headScheduler: Scheduler
 ) : ApplicationRunner {
    @Value("\${spring.application.max-metadata-size}")
    private var maxMetadataSize: Int = Defaults.maxMetadataSize
@@ -261,11 +263,11 @@ open class ConfiguredUpstreams(
        }

        val extractBlock = ExtractBlock()
-        val rpcHead = BitcoinRpcHead(directApi, extractBlock)
+        val rpcHead = BitcoinRpcHead(directApi, extractBlock, headScheduler = headScheduler)
        val head: Head = conn.zeroMq?.let { zeroMq ->
            val server = ZMQServer(zeroMq.host, zeroMq.port, "hashblock")
-            val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock)
-            MergedHead(listOf(rpcHead, zeroMqHead), MostWorkForkChoice())
+            val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock, headScheduler)
+            MergedHead(listOf(rpcHead, zeroMqHead), MostWorkForkChoice(), headScheduler)
        } ?: rpcHead

        val methods = buildMethods(config, chain)
@@ -350,7 +352,8 @@ open class ConfiguredUpstreams(
            chainsConfig,
            grpcTracing,
            clientSpansInterceptor,
-            maxMetadataSize
+            maxMetadataSize,
+            headScheduler
        ).apply {
            timeout = options.timeout
        }
@@ -412,7 +415,13 @@ open class ConfiguredUpstreams(
        log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
        val connectorFactory =
            EthereumConnectorFactory(
-                conn.resolveMode(), wsFactoryApi, httpFactory, forkChoice, blockValidator, wsConnectionResubscribeScheduler
+                conn.resolveMode(),
+                wsFactoryApi,
+                httpFactory,
+                forkChoice,
+                blockValidator,
+                wsConnectionResubscribeScheduler,
+                headScheduler
            )
        if (!connectorFactory.isValid()) {
            log.warn("Upstream configuration is invalid (probably no http endpoint)")
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt
@@ -26,7 +26,7 @@ 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.core.scheduler.Scheduler
 import reactor.kotlin.core.publisher.toMono
 import java.util.concurrent.Executors
 import java.util.concurrent.Future
@@ -37,9 +37,10 @@ import java.util.concurrent.locks.ReentrantLock

 abstract class AbstractHead @JvmOverloads constructor(
    private val forkChoice: ForkChoice,
+    private val headScheduler: Scheduler,
    private val blockValidator: BlockValidator = BlockValidator.ALWAYS_VALID,
    private val awaitHeadTimeoutMs: Long = 60_000,
-    private val upstreamId: String = ""
+    private val upstreamId: String = "",
 ) : Head {
    protected val log = LoggerFactory.getLogger(this::class.java)

@@ -83,7 +84,7 @@ abstract class AbstractHead @JvmOverloads constructor(
                    log.warn("Received signal $upstreamId $it, continue emit heads")
                }
            }
-            .subscribeOn(Schedulers.boundedElastic())
+            .subscribeOn(headScheduler)
            .subscribe { block ->
                val valid = runCatching {
                    blockValidator.isValid(forkChoice.getHead(), block)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt
@@ -9,11 +9,11 @@ import reactor.core.scheduler.Scheduler
 open class DynamicMergedHead(
    forkChoice: ForkChoice,
    private val label: String = "",
-    scheduler: Scheduler
-) : AbstractHead(forkChoice, upstreamId = label), Lifecycle {
+    headScheduler: Scheduler
+) : AbstractHead(forkChoice, headScheduler, upstreamId = label), Lifecycle {

    private var subscription: Disposable? = null
-    private val dynamicFlux: DynamicMergeFlux<String, BlockContainer> = DynamicMergeFlux(scheduler)
+    private val dynamicFlux: DynamicMergeFlux<String, BlockContainer> = DynamicMergeFlux(headScheduler)

    override fun isRunning(): Boolean {
        return subscription != null
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt
@@ -21,7 +21,7 @@ import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
-import reactor.core.scheduler.Schedulers
+import reactor.core.scheduler.Scheduler
 import reactor.util.function.Tuple2
 import reactor.util.function.Tuples
 import java.time.Duration
@@ -35,6 +35,7 @@ abstract class HeadLagObserver(
    private val master: Head,
    private val followers: Collection<Upstream>,
    private val distanceExtractor: Extractor,
+    private val lagObserverScheduler: Scheduler,
    private val throttling: Duration = Duration.ofSeconds(5)
 ) : Lifecycle {

@@ -44,7 +45,7 @@ abstract class HeadLagObserver(

    override fun start() {
        current?.dispose()
-        current = subscription().subscribe { }
+        current = subscription().subscribeOn(lagObserverScheduler).subscribe { }
    }

    override fun isRunning(): Boolean {
@@ -68,7 +69,7 @@ abstract class HeadLagObserver(
    fun probeFollowers(top: BlockContainer): Flux<Tuple2<Long, Upstream>> {
        return Flux.fromIterable(followers)
            .parallel(followers.size)
-            .flatMap { up -> mapLagging(top, up, getCurrentBlocks(up)).subscribeOn(Schedulers.boundedElastic()) }
+            .flatMap { up -> mapLagging(top, up, getCurrentBlocks(up)).subscribeOn(lagObserverScheduler) }
            .sequential()
            .onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) }
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt
@@ -22,12 +22,14 @@ import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
+import reactor.core.scheduler.Scheduler

 class MergedHead @JvmOverloads constructor(
    private val sources: Iterable<Head>,
    forkChoice: ForkChoice,
+    headScheduler: Scheduler,
    private val label: String = ""
-) : AbstractHead(forkChoice, upstreamId = label), Lifecycle, CachesEnabled {
+) : AbstractHead(forkChoice, headScheduler, upstreamId = label), Lifecycle, CachesEnabled {

    private var subscription: Disposable? = null

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt
@@ -21,11 +21,13 @@ import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.HeadLagObserver
 import io.emeraldpay.dshackle.upstream.Upstream
 import org.slf4j.LoggerFactory
+import reactor.core.scheduler.Scheduler

 class BitcoinHeadLagObserver(
    master: Head,
-    followers: Collection<Upstream>
-) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) {
+    followers: Collection<Upstream>,
+    headScheduler: Scheduler
+) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance, headScheduler) {

    companion object {
        private val log = LoggerFactory.getLogger(BitcoinHeadLagObserver::class.java)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
@@ -33,12 +33,14 @@ import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler

@Suppress("UNCHECKED_CAST")
 open class BitcoinMultistream(
    chain: Chain,
    private val sourceUpstreams: MutableList<BitcoinUpstream>,
    caches: Caches,
+    private val headScheduler: Scheduler,
 ) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, caches), Lifecycle {

    private var head: Head = EmptyHead()
@@ -86,7 +88,7 @@ open class BitcoinMultistream(
                }
            }
        } else {
-            val newHead = MergedHead(sourceUpstreams.map { it.getHead() }, MostWorkForkChoice()).apply {
+            val newHead = MergedHead(sourceUpstreams.map { it.getHead() }, MostWorkForkChoice(), headScheduler).apply {
            this.start()
        }
            newHead
@@ -152,7 +154,7 @@ open class BitcoinMultistream(
    }

    override fun makeLagObserver(): HeadLagObserver {
-        return BitcoinHeadLagObserver(head, sourceUpstreams)
+        return BitcoinHeadLagObserver(head, sourceUpstreams, headScheduler)
    }

    override fun start() {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
@@ -27,6 +27,7 @@ import org.springframework.scheduling.concurrent.CustomizableThreadFactory
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
 import reactor.core.scheduler.Schedulers
 import java.time.Duration
 import java.util.concurrent.Executors
@@ -34,8 +35,9 @@ import java.util.concurrent.Executors
 class BitcoinRpcHead(
    private val api: JsonRpcReader,
    private val extractBlock: ExtractBlock,
-    private val interval: Duration = Duration.ofSeconds(15)
-) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle {
+    private val interval: Duration = Duration.ofSeconds(15),
+    headScheduler: Scheduler
+) : Head, AbstractHead(MostWorkForkChoice(), headScheduler, awaitHeadTimeoutMs = 1200_000), Lifecycle {

    companion object {
        val scheduler =
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
@@ -49,13 +49,6 @@ open class BitcoinRpcUpstream(
        setOf(Capability.RPC)
    }

-    private fun createHead(): Head {
-        return BitcoinRpcHead(
-            directApi,
-            ExtractBlock()
-        )
-    }
-
    override fun getHead(): Head {
        return head
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
@@ -13,6 +13,7 @@ import org.apache.commons.codec.binary.Hex
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
 import reactor.util.retry.Retry
 import java.time.Duration

@@ -20,7 +21,8 @@ class BitcoinZMQHead(
    private val server: ZMQServer,
    private val api: JsonRpcReader,
    private val extractBlock: ExtractBlock,
-) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle {
+    headScheduler: Scheduler,
+) : Head, AbstractHead(MostWorkForkChoice(), headScheduler, awaitHeadTimeoutMs = 1200_000), Lifecycle {
    private var refreshSubscription: Disposable? = null

    fun connect(): Flux<BlockContainer> {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
@@ -25,16 +25,18 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.etherjar.hex.HexQuantity
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler

 open class DefaultEthereumHead(
    protected val upstreamId: String,
    forkChoice: ForkChoice,
-    blockValidator: BlockValidator
-) : Head, AbstractHead(forkChoice, blockValidator, 60_000, upstreamId) {
+    blockValidator: BlockValidator,
+    private val headScheduler: Scheduler,
+) : Head, AbstractHead(forkChoice, headScheduler, blockValidator, 60_000, upstreamId) {

    fun getLatestBlock(api: JsonRpcReader): Mono<BlockContainer> {
        return api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
-            .subscribeOn(EthereumRpcHead.scheduler)
+            .subscribeOn(headScheduler)
            .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
            .flatMap {
                if (it.error != null) {
@@ -48,7 +50,7 @@ open class DefaultEthereumHead(
                // fetching by Block Height here, critical to use the same upstream as in previous call,
                // b/c different upstreams may have different blocks on the same height
                api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
-                    .subscribeOn(EthereumRpcHead.scheduler)
+                    .subscribeOn(headScheduler)
                    .timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
            }
            .map {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt
@@ -22,11 +22,13 @@ import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.HeadLagObserver
 import io.emeraldpay.dshackle.upstream.Upstream
 import org.slf4j.LoggerFactory
+import reactor.core.scheduler.Scheduler

 class EthereumHeadLagObserver(
    master: Head,
-    followers: Collection<Upstream>
-) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) {
+    followers: Collection<Upstream>,
+    headScheduler: Scheduler
+) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance, headScheduler) {

    companion object {
        private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
@@ -49,8 +49,8 @@ open class EthereumMultistream(
    chain: Chain,
    val upstreams: MutableList<EthereumUpstream>,
    caches: Caches,
-    headScheduler: Scheduler,
-    tracer: Tracer
+    private val headScheduler: Scheduler,
+    tracer: Tracer,
 ) : Multistream(chain, upstreams as MutableList<Upstream>, caches), EthereumLikeMultistream {

    private var head: DynamicMergedHead = DynamicMergedHead(
@@ -131,7 +131,7 @@ open class EthereumMultistream(
    }

    override fun makeLagObserver(): HeadLagObserver {
-        return EthereumHeadLagObserver(head, upstreams as Collection<Upstream>)
+        return EthereumHeadLagObserver(head, upstreams as Collection<Upstream>, headScheduler)
    }

    override fun isRunning(): Boolean {
@@ -192,7 +192,7 @@ open class EthereumMultistream(
                    when (it.size) {
                        0 -> EmptyHead()
                        1 -> selected.first()
-                        else -> MergedHead(selected, MostWorkForkChoice(), "Eth head ${it.map { it.getId() }}").apply {
+                        else -> MergedHead(selected, MostWorkForkChoice(), headScheduler, "Eth head ${it.map { it.getId() }}").apply {
                            start()
                        }
                    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
@@ -20,25 +20,19 @@ import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
-import org.springframework.scheduling.concurrent.CustomizableThreadFactory
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
-import reactor.core.scheduler.Schedulers
+import reactor.core.scheduler.Scheduler
 import java.time.Duration
-import java.util.concurrent.Executors

 class EthereumRpcHead(
    private val api: JsonRpcReader,
    forkChoice: ForkChoice,
    upstreamId: String,
    blockValidator: BlockValidator,
+    private val headScheduler: Scheduler,
    private val interval: Duration = Duration.ofSeconds(10),
-) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {
-
-    companion object {
-        val scheduler =
-            Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("ethereum-rpc-head")))
-    }
+) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator, headScheduler), Lifecycle {

    private var refreshSubscription: Disposable? = null

@@ -46,7 +40,7 @@ class EthereumRpcHead(
        super.start()
        refreshSubscription?.dispose()
        val base = Flux.interval(interval)
-            .publishOn(scheduler)
+            .publishOn(headScheduler)
            .flatMap {
                getLatestBlock(api)
            }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
@@ -44,8 +44,9 @@ class EthereumWsHead(
    private val api: JsonRpcReader,
    private val wsSubscriptions: WsSubscriptions,
    private val skipEnhance: Boolean,
-    private val wsConnectionResubscribeScheduler: Scheduler
-) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {
+    private val wsConnectionResubscribeScheduler: Scheduler,
+    headScheduler: Scheduler,
+) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator, headScheduler), Lifecycle {

    private var connectionId: String? = null
    private var subscribed = false
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt
@@ -313,7 +313,7 @@ open class WsConnectionImpl(
        )
        val sender = currentRequests.remove(msg.id.asNumber().toInt())
        if (sender == null) {
-            log.warn("Unknown response received for ${msg.id}")
+            log.warn("Unknown response received for ${msg.id} with body ${msg.value?.let { String(it) }}")
        } else {
            try {
                val emitResult = sender.tryEmitValue(rpcResponse)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
@@ -19,7 +19,8 @@ open class EthereumConnectorFactory(
    private val httpFactory: HttpFactory?,
    private val forkChoice: ForkChoice,
    private val blockValidator: BlockValidator,
-    private val wsConnectionResubscribeScheduler: Scheduler
+    private val wsConnectionResubscribeScheduler: Scheduler,
+    private val headScheduler: Scheduler
 ) : ConnectorFactory {

    override fun isValid(): Boolean {
@@ -50,7 +51,13 @@ open class EthereumConnectorFactory(
    ): EthereumConnector {
        if (wsFactory != null && connectorType == WS_ONLY) {
            return EthereumWsConnector(
-                wsFactory, upstream, forkChoice, blockValidator, skipEnhance, wsConnectionResubscribeScheduler
+                wsFactory,
+                upstream,
+                forkChoice,
+                blockValidator,
+                skipEnhance,
+                wsConnectionResubscribeScheduler,
+                headScheduler
            )
        }
        if (httpFactory == null) {
@@ -64,7 +71,8 @@ open class EthereumConnectorFactory(
            forkChoice,
            blockValidator,
            skipEnhance,
-            wsConnectionResubscribeScheduler
+            wsConnectionResubscribeScheduler,
+            headScheduler
        )
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
@@ -33,7 +33,8 @@ class EthereumRpcConnector(
    forkChoice: ForkChoice,
    blockValidator: BlockValidator,
    skipEnhance: Boolean,
-    wsConnectionResubscribeScheduler: Scheduler
+    wsConnectionResubscribeScheduler: Scheduler,
+    headScheduler: Scheduler
 ) : EthereumConnector, CachesEnabled {
    private val pool: WsConnectionPool?
    private val head: Head
@@ -43,35 +44,50 @@ class EthereumRpcConnector(
    }

    init {
-        if (wsFactory != null) {
-            pool = wsFactory.create(null)
-        } else {
-            pool = null
-        }
+        pool = wsFactory?.create(null)

        head = when (connectorType) {
            RPC_ONLY -> {
                log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC")
-                EthereumRpcHead(getIngressReader(), forkChoice, id, blockValidator)
+                EthereumRpcHead(getIngressReader(), forkChoice, id, blockValidator, headScheduler)
            }
+
            WS_ONLY -> {
                throw IllegalStateException("WS-only mode is not supported in RPC connector")
            }
+
            RPC_REQUESTS_WITH_MIXED_HEAD -> {
                val wsHead =
                    EthereumWsHead(
-                        id, AlwaysForkChoice(), blockValidator, getIngressReader(),
-                        WsSubscriptionsImpl(pool!!), skipEnhance, wsConnectionResubscribeScheduler
+                        id,
+                        AlwaysForkChoice(),
+                        blockValidator,
+                        getIngressReader(),
+                        WsSubscriptionsImpl(pool!!),
+                        skipEnhance,
+                        wsConnectionResubscribeScheduler,
+                        headScheduler
                    )
                // receive all new blocks through WebSockets, but also periodically verify with RPC in case if WS failed
                val rpcHead =
-                    EthereumRpcHead(getIngressReader(), AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
-                MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id")
+                    EthereumRpcHead(
+                        getIngressReader(),
+                        AlwaysForkChoice(),
+                        id,
+                        blockValidator,
+                        headScheduler,
+                        Duration.ofSeconds(30)
+                    )
+                MergedHead(listOf(rpcHead, wsHead), forkChoice, headScheduler, "Merged for $id")
            }
+
            RPC_REQUESTS_WITH_WS_HEAD -> {
                EthereumWsHead(
-                    id, AlwaysForkChoice(), blockValidator, getIngressReader(),
-                    WsSubscriptionsImpl(pool!!), skipEnhance, wsConnectionResubscribeScheduler
+                    id,
+                    AlwaysForkChoice(),
+                    blockValidator, getIngressReader(),
+                    WsSubscriptionsImpl(pool!!), skipEnhance, wsConnectionResubscribeScheduler,
+                    headScheduler
                )
            }
        }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
@@ -20,7 +20,8 @@ class EthereumWsConnector(
    forkChoice: ForkChoice,
    blockValidator: BlockValidator,
    skipEnhance: Boolean,
-    wsConnectionResubscribeScheduler: Scheduler
+    wsConnectionResubscribeScheduler: Scheduler,
+    headScheduler: Scheduler
 ) : EthereumConnector {
    private val pool: WsConnectionPool
    private val reader: JsonRpcReader
@@ -32,8 +33,14 @@ class EthereumWsConnector(
        reader = JsonRpcWsClient(pool)
        val wsSubscriptions = WsSubscriptionsImpl(pool)
        head = EthereumWsHead(
-            upstream.getId(), forkChoice, blockValidator, reader,
-            wsSubscriptions, skipEnhance, wsConnectionResubscribeScheduler
+            upstream.getId(),
+            forkChoice,
+            blockValidator,
+            reader,
+            wsSubscriptions,
+            skipEnhance,
+            wsConnectionResubscribeScheduler,
+            headScheduler
        )
        subscriptions = EthereumWsIngressSubscription(wsSubscriptions)
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt
@@ -6,11 +6,13 @@ import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.HeadLagObserver
 import io.emeraldpay.dshackle.upstream.Upstream
 import org.slf4j.LoggerFactory
+import reactor.core.scheduler.Scheduler

 class EthereumPosHeadLagObserver(
    master: Head,
-    followers: Collection<Upstream>
-) : HeadLagObserver(master, followers, DistanceExtractor::extractPriorityDistance) {
+    followers: Collection<Upstream>,
+    headScheduler: Scheduler
+) : HeadLagObserver(master, followers, DistanceExtractor::extractPriorityDistance, headScheduler) {

    companion object {
        private val log = LoggerFactory.getLogger(EthereumPosHeadLagObserver::class.java)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
@@ -47,7 +47,7 @@ open class EthereumPosMultiStream(
    chain: Chain,
    val upstreams: MutableList<EthereumPosUpstream>,
    caches: Caches,
-    headScheduler: Scheduler,
+    private val headScheduler: Scheduler,
    tracer: Tracer
 ) : Multistream(chain, upstreams as MutableList<Upstream>, caches), EthereumLikeMultistream {

@@ -104,7 +104,7 @@ open class EthereumPosMultiStream(
    }

    override fun makeLagObserver(): HeadLagObserver =
-        EthereumPosHeadLagObserver(head, ArrayList(upstreams)).apply {
+        EthereumPosHeadLagObserver(head, ArrayList(upstreams), headScheduler).apply {
            start()
        }

@@ -169,6 +169,7 @@ open class EthereumPosMultiStream(
                            else -> MergedHead(
                                selected,
                                PriorityForkChoice(),
+                                headScheduler,
                                "ETH head for ${it.map { it.getId() }}"
                            ).apply {
                                start()
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
@@ -40,6 +40,7 @@ import io.emeraldpay.etherjar.rpc.RpcException
 import org.reactivestreams.Publisher
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
 import java.math.BigInteger
 import java.time.Instant
 import java.util.Locale
@@ -51,9 +52,10 @@ class BitcoinGrpcUpstream(
    role: UpstreamsConfig.UpstreamRole,
    chain: Chain,
    val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
-    private val client: JsonRpcGrpcClient,
+    client: JsonRpcGrpcClient,
    overrideLabels: UpstreamsConfig.Labels?,
-    chainConfig: ChainsConfig.ChainConfig
+    chainConfig: ChainsConfig.ChainConfig,
+    headScheduler: Scheduler,
 ) : BitcoinUpstream(
    "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
    chain,
@@ -100,7 +102,10 @@ class BitcoinGrpcUpstream(
            }
    }
    private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
-    private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
+    private val grpcHead = GrpcHead(
+        getId(), chain, this, remote, blockConverter, reloadBlock,
+        MostWorkForkChoice(), headScheduler
+    )
    private val timeout = Defaults.timeout
    private var capabilities: Set<Capability> = emptySet()
    private val buildInfo: BuildInfo = BuildInfo()
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
@@ -45,6 +45,7 @@ import io.emeraldpay.etherjar.rpc.RpcException
 import org.reactivestreams.Publisher
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
 import java.math.BigInteger
 import java.time.Instant
 import java.util.Locale
@@ -59,7 +60,8 @@ open class EthereumGrpcUpstream(
    private val remote: ReactorBlockchainStub,
    client: JsonRpcGrpcClient,
    overrideLabels: UpstreamsConfig.Labels?,
-    chainConfig: ChainsConfig.ChainConfig
+    chainConfig: ChainsConfig.ChainConfig,
+    headScheduler: Scheduler,
 ) : EthereumUpstream(
    "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
    hash,
@@ -110,7 +112,10 @@ open class EthereumGrpcUpstream(
    }

    private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
-    private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
+    private val grpcHead = GrpcHead(
+        getId(), chain, this, remote,
+        blockConverter, reloadBlock, MostWorkForkChoice(), headScheduler
+    )
    private var capabilities: Set<Capability> = emptySet()
    private val buildInfo: BuildInfo = BuildInfo()

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
@@ -39,6 +39,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.etherjar.domain.BlockHash
 import reactor.core.publisher.Flux
+import reactor.core.scheduler.Scheduler
 import java.math.BigInteger
 import java.time.Instant
 import java.util.Locale
@@ -53,7 +54,8 @@ open class EthereumPosGrpcUpstream(
    client: JsonRpcGrpcClient,
    nodeRating: Int,
    overrideLabels: UpstreamsConfig.Labels?,
-    chainConfig: ChainsConfig.ChainConfig
+    chainConfig: ChainsConfig.ChainConfig,
+    headScheduler: Scheduler,
 ) : EthereumPosUpstream(
    "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
    hash,
@@ -84,7 +86,10 @@ open class EthereumPosGrpcUpstream(
    }

    private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
-    private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, null, NoChoiceWithPriorityForkChoice(nodeRating, parentId))
+    private val grpcHead = GrpcHead(
+        getId(), chain, this, remote, blockConverter, null,
+        NoChoiceWithPriorityForkChoice(nodeRating, parentId), headScheduler
+    )
    private var capabilities: Set<Capability> = emptySet()
    private val buildInfo: BuildInfo = BuildInfo()

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
@@ -31,6 +31,7 @@ import org.reactivestreams.Publisher
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
 import reactor.kotlin.extra.retry.retryExponentialBackoff
 import java.time.Duration
 import java.util.function.Function
@@ -48,8 +49,9 @@ class GrpcHead(
     * Populate block data with all missing details, of any
     */
    private val enhancer: Function<BlockContainer, Publisher<BlockContainer>>?,
-    private val forkChoice: ForkChoice
-) : AbstractHead(forkChoice, upstreamId = id), Lifecycle {
+    private val forkChoice: ForkChoice,
+    headScheduler: Scheduler,
+) : AbstractHead(forkChoice, headScheduler, upstreamId = id), Lifecycle {

    private var headSubscription: Disposable? = null

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
@@ -75,7 +75,8 @@ class GrpcUpstreams(
    private val chainsConfig: ChainsConfig,
    private val grpcTracing: GrpcTracing,
    private val clientSpansInterceptor: ClientInterceptor?,
-    private var maxMetadataSize: Int
+    private var maxMetadataSize: Int,
+    private val headScheduler: Scheduler
 ) {
    private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)

@@ -213,13 +214,34 @@ class GrpcUpstreams(

    private val creators: Map<BlockchainType, (chain: Chain, client: JsonRpcGrpcClient) -> DefaultUpstream> = mapOf(
        BlockchainType.EVM_POW to { chain, rpcClient ->
-            EthereumGrpcUpstream(id, hash, role, chain, client, rpcClient, labels, chainsConfig.resolve(chain))
+            EthereumGrpcUpstream(
+                id,
+                hash,
+                role,
+                chain,
+                client,
+                rpcClient,
+                labels,
+                chainsConfig.resolve(chain),
+                headScheduler
+            )
        },
        BlockchainType.EVM_POS to { chain, rpcClient ->
-            EthereumPosGrpcUpstream(id, hash, role, chain, client, rpcClient, nodeRating, labels, chainsConfig.resolve(chain))
+            EthereumPosGrpcUpstream(
+                id,
+                hash,
+                role,
+                chain,
+                client,
+                rpcClient,
+                nodeRating,
+                labels,
+                chainsConfig.resolve(chain),
+                headScheduler
+            )
        },
        BlockchainType.BITCOIN to { chain, rpcClient ->
-            BitcoinGrpcUpstream(id, role, chain, client, rpcClient, labels, chainsConfig.resolve(chain))
+            BitcoinGrpcUpstream(id, role, chain, client, rpcClient, labels, chainsConfig.resolve(chain), headScheduler)
        }
    )