diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt index 59e0665c..fdf94a89 100644 --- 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 { 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) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt index 5f54f276..1b6ebc9b 100644 --- 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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index c442cab0..c46d1fb5 100644 --- 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)") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 4beeb55c..64171e72 100644 --- 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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt index e5f957ec..175ba847 100644 --- 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 = DynamicMergeFlux(scheduler) + private val dynamicFlux: DynamicMergeFlux = DynamicMergeFlux(headScheduler) override fun isRunning(): Boolean { return subscription != null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt index 054c4247..bc2224db 100644 --- 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, 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> { 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) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt index 6c4ab0ac..6cfb7f3f 100644 --- 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, 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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt index fbfb663d..b7abae4e 100644 --- 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 -) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) { + followers: Collection, + headScheduler: Scheduler +) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance, headScheduler) { companion object { private val log = LoggerFactory.getLogger(BitcoinHeadLagObserver::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index 2495118a..d41b81cb 100644 --- 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, caches: Caches, + private val headScheduler: Scheduler, ) : Multistream(chain, sourceUpstreams as MutableList, 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() { 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 68067898..52f9ca98 100644 --- 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 = diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt index e013132c..cb6e4f6e 100644 --- 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 } 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 7a13dff2..9451eb61 100644 --- 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt index 5725ea06..f1d6572b 100644 --- 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 { 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt index c933214b..61e7eb2f 100644 --- 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 -) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) { + followers: Collection, + headScheduler: Scheduler +) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance, headScheduler) { companion object { private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index d082100a..19103a9e 100644 --- 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, caches: Caches, - headScheduler: Scheduler, - tracer: Tracer + private val headScheduler: Scheduler, + tracer: Tracer, ) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { private var head: DynamicMergedHead = DynamicMergedHead( @@ -131,7 +131,7 @@ open class EthereumMultistream( } override fun makeLagObserver(): HeadLagObserver { - return EthereumHeadLagObserver(head, upstreams as Collection) + return EthereumHeadLagObserver(head, upstreams as Collection, 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() } } 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 2c4cc9ca..193c833b 100644 --- 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) } 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 619500c6..d791e40b 100644 --- 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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt index 7723b3ef..e18f31f6 100644 --- 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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt index 7a032c8b..a03e2e0e 100644 --- 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 ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index eaf8f1b9..fa11a62a 100644 --- 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 ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index 05ff6cb4..2239772a 100644 --- 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) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt index 4b44c58c..ff806f19 100644 --- 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 -) : HeadLagObserver(master, followers, DistanceExtractor::extractPriorityDistance) { + followers: Collection, + headScheduler: Scheduler +) : HeadLagObserver(master, followers, DistanceExtractor::extractPriorityDistance, headScheduler) { companion object { private val log = LoggerFactory.getLogger(EthereumPosHeadLagObserver::class.java) 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 67b31d89..75f0e3f6 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 @@ -47,7 +47,7 @@ open class EthereumPosMultiStream( chain: Chain, val upstreams: MutableList, caches: Caches, - headScheduler: Scheduler, + private val headScheduler: Scheduler, tracer: Tracer ) : Multistream(chain, upstreams as MutableList, 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() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index fa408240..23e1bccf 100644 --- 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 = emptySet() private val buildInfo: BuildInfo = BuildInfo() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 2c430e93..f52783d2 100644 --- 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 = emptySet() private val buildInfo: BuildInfo = BuildInfo() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index fa9c3ef3..72c9e771 100644 --- 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 = emptySet() private val buildInfo: BuildInfo = BuildInfo() 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 7cf6e2d7..d7bc61f7 100644 --- 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>?, - 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 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 50e575c8..ecdc9a30 100644 --- 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 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) } ) diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index 648f375c..ce70a395 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -31,7 +31,8 @@ class ConfiguredUpstreamsSpec extends Specification { ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), Schedulers.boundedElastic(), - null + null, + Schedulers.boundedElastic(), ) def methods = new UpstreamsConfig.Methods( [ @@ -62,7 +63,8 @@ class ConfiguredUpstreamsSpec extends Specification { ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), Schedulers.boundedElastic(), - null + null, + Schedulers.boundedElastic(), ) def methods = new UpstreamsConfig.Methods( [ @@ -92,7 +94,8 @@ class ConfiguredUpstreamsSpec extends Specification { ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), Schedulers.boundedElastic(), - null + null, + Schedulers.boundedElastic(), ) expect: configurer.getHash(node, src) == expected @@ -117,7 +120,8 @@ class ConfiguredUpstreamsSpec extends Specification { ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), Schedulers.boundedElastic(), - null + null, + Schedulers.boundedElastic(), ) when: def h1 = configurer.getHash(null, "hohoho") @@ -147,7 +151,8 @@ class ConfiguredUpstreamsSpec extends Specification { ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), Schedulers.boundedElastic(), - null + null, + Schedulers.boundedElastic(), ) def methodsGroup = new UpstreamsConfig.MethodGroups( ["filter"] as Set, diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy index 53add366..aa980a7a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy @@ -22,6 +22,7 @@ import org.jetbrains.annotations.NotNull import reactor.core.publisher.Sinks import reactor.test.StepVerifier import spock.lang.Specification +import reactor.core.scheduler.Schedulers import java.time.Duration import java.time.Instant @@ -137,7 +138,7 @@ class AbstractHeadSpec extends Specification { BlockContainer getHead() { return null } - }, new BlockValidator.AlwaysValid(), 100_000) + }, Schedulers.boundedElastic(), new BlockValidator.AlwaysValid() , 100_000) } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 0f778381..93bb7ea1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -51,12 +51,17 @@ class FilteredApisSpec extends Specification { create(_, _) >> TestingCommons.api().tap { it.id = "${i++}" } } def connectorFactory = new EthereumConnectorFactory( - EthereumConnectorFactory.ConnectorMode.RPC_ONLY, null, httpFactory, - new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.boundedElastic() + EthereumConnectorFactory.ConnectorMode.RPC_ONLY, + null, + httpFactory, + new MostWorkForkChoice(), + BlockValidator.ALWAYS_VALID, + Schedulers.boundedElastic(), + Schedulers.boundedElastic() ) new EthereumRpcUpstream( "test", - (byte)123, + (byte) 123, Chain.ETHEREUM, new UpstreamsConfig.PartialOptions().buildOptions(), UpstreamsConfig.UpstreamRole.PRIMARY, @@ -79,8 +84,8 @@ class FilteredApisSpec extends Specification { .expectNext(upstreams[0]) .expectNext(upstreams[2]) .expectNext(upstreams[3]) - .expectComplete() - .verify(Duration.ofSeconds(1)) + .expectComplete() + .verify(Duration.ofSeconds(1)) when: iter = new FilteredApis(Chain.ETHEREUM, upstreams, matcher, 1, 1, 0) @@ -114,19 +119,19 @@ class FilteredApisSpec extends Specification { expect: wait == apis.waitDuration(n).toMillis() as Integer where: - n | wait - 0 | 100 - 1 | 100 - 2 | 400 - 3 | 900 - 4 | 1600 - 5 | 2500 - 6 | 3600 - 7 | 4900 - 8 | 5000 - 9 | 5000 - 10 | 5000 - -1 | 100 + n | wait + 0 | 100 + 1 | 100 + 2 | 400 + 3 | 900 + 4 | 1600 + 5 | 2500 + 6 | 3600 + 7 | 4900 + 8 | 5000 + 9 | 5000 + 10 | 5000 + -1 | 100 } @Retry @@ -169,8 +174,8 @@ class FilteredApisSpec extends Specification { .expectNext(up1, up2).as("Batch 3") .expectNoEvent(Duration.ofMillis(900)).as("Wait 3") .expectNext(up1, up2).as("Batch 4") - .expectComplete() - .verify(Duration.ofSeconds(10)) + .expectComplete() + .verify(Duration.ofSeconds(10)) } def "Starts with right position"() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy index 090e043c..2abf7a0c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy @@ -22,10 +22,12 @@ import io.emeraldpay.etherjar.rpc.json.BlockJson import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux import reactor.core.publisher.Sinks +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import reactor.util.function.Tuples import spock.lang.Specification + import java.time.Duration import java.time.Instant @@ -114,7 +116,8 @@ class HeadLagObserverSpec extends Specification { class TestHeadLagObserver extends HeadLagObserver { TestHeadLagObserver(@NotNull Head master, @NotNull Collection followers) { - super(master, followers, DistanceExtractor.@Companion::extractPowDistance, Duration.ofNanos(1)) + super(master, followers, DistanceExtractor.@Companion::extractPowDistance, + Schedulers.boundedElastic(), Duration.ofNanos(1)) } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy index 4bcd55d3..3d84b519 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers import spock.lang.Specification class MergedHeadSpec extends Specification { @@ -36,7 +37,7 @@ class MergedHeadSpec extends Specification { } when: - def merged = new MergedHead([head1, head2, head3], new MostWorkForkChoice()) + def merged = new MergedHead([head1, head2, head3], new MostWorkForkChoice(), Schedulers.boundedElastic()) merged.start() then: @@ -45,14 +46,14 @@ class MergedHeadSpec extends Specification { class TestHead1 extends AbstractHead { TestHead1() { - super(new MostWorkForkChoice(), new BlockValidator.AlwaysValid(), 100_000) + super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000) } } class TestHead2 extends AbstractHead implements Lifecycle { TestHead2() { - super(new MostWorkForkChoice(), new BlockValidator.AlwaysValid(), 100_000) + super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000) } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy index 08bb96d5..45a1cebf 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy @@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -85,7 +86,7 @@ class BitcoinRpcHeadSpec extends Specification { _ * read(new JsonRpcRequest("getblock", [hash1])) >> Mono.just(new JsonRpcResponse(block1.bytes, null)) _ * read(new JsonRpcRequest("getblock", [hash2])) >> Mono.just(new JsonRpcResponse(block2.bytes, null)) } - BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(), Duration.ofMillis(200)) + BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(), Duration.ofMillis(200), Schedulers.boundedElastic()) when: def act = head.flux.take(2) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy index c5acc552..e7cc8a9d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.rpc.json.BlockJson import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -31,7 +32,7 @@ import java.time.Instant class DefaultEthereumHeadSpec extends Specification { - DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID) + DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.boundedElastic()) ObjectMapper objectMapper = Global.objectMapper BlockHash parent = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915210") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy index c5c5f936..aa0d3503 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy @@ -66,7 +66,7 @@ class EthereumWsHeadSpec extends Specification { 1 * it.connectionInfoFlux() >> Flux.empty() } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.boundedElastic(), Schedulers.boundedElastic()) when: def act = head.listenNewHeads().blockFirst() @@ -107,7 +107,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) when: def act = head.getFlux() @@ -161,7 +161,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) when: def act = head.getFlux() @@ -201,7 +201,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) when: def act = head.getFlux() @@ -240,7 +240,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) when: def act = head.getFlux() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index c276ae1d..a1d00114 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -37,6 +37,7 @@ import io.emeraldpay.etherjar.rpc.json.BlockJson import io.grpc.stub.StreamObserver import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Timer +import reactor.core.scheduler.Schedulers import spock.lang.Specification import java.time.Duration @@ -89,7 +90,7 @@ class EthereumGrpcUpstreamSpec extends Specification { ) } }) - def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default()) + def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic()) upstream.setLag(0) upstream.update( BlockchainOuterClass.DescribeChain.newBuilder() @@ -160,7 +161,7 @@ class EthereumGrpcUpstreamSpec extends Specification { }).start() } }) - def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null, ChainsConfig.ChainConfig.default()) + def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic()) upstream.setLag(0) upstream.update( BlockchainOuterClass.DescribeChain.newBuilder() @@ -232,7 +233,7 @@ class EthereumGrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default()) + def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic()) upstream.setLag(0) upstream.update( BlockchainOuterClass.DescribeChain.newBuilder() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index 2ebec4c3..99b3a128 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.grpc.stub.StreamObserver +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -66,7 +67,10 @@ class GrpcHeadSpec extends Specification { Chain.BITCOIN, Stub(DefaultUpstream), client, - convert, null, new MostWorkForkChoice() + convert, + null, + new MostWorkForkChoice(), + Schedulers.boundedElastic() ) when: def act = head.getFlux() @@ -130,7 +134,10 @@ class GrpcHeadSpec extends Specification { Chain.BITCOIN, Stub(DefaultUpstream), client, - convert, null, new MostWorkForkChoice() + convert, + null, + new MostWorkForkChoice(), + Schedulers.boundedElastic() ) when: def act = head.getFlux()