diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt new file mode 100644 index 00000000..43552c24 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt @@ -0,0 +1,94 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import com.google.common.cache.CacheBuilder +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle +import io.emeraldpay.etherjar.domain.BlockHash +import reactor.core.Disposable +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.publisher.Sinks +import reactor.core.scheduler.Scheduler +import java.time.Duration + +class EnrichedMergedHead constructor( + private val sources: Iterable, + private val referenceHead: Head, + private val headScheduler: Scheduler, + private val api: Reader +) : Head, Lifecycle { + + private val enrichedBlocks = CacheBuilder.newBuilder() + .maximumSize(10) + .build() + private val enrichedPromises = CacheBuilder.newBuilder() + .maximumSize(10) + .build>() + private var cacheSub: Disposable? = null + + private fun getEnrichBlockMono(id: BlockId): Mono { + val block = enrichedBlocks.getIfPresent(id) + return if (block != null) { + Mono.just(block) + } else { + enrichedPromises.get(id) { + Sinks.one() + }.asMono() + } + } + + override fun getFlux(): Flux { + return referenceHead.getFlux().concatMap { block -> + if (block.enriched) { + Mono.just(block) + } else { + Mono.firstWithValue( + getEnrichBlockMono(block.hash), + Mono.just(block) + .delayElement(Duration.ofSeconds(1)) + .flatMap { + EthereumBlockEnricher.enrich(BlockHash(block.hash.value), api, headScheduler) + } + ) + } + } + } + + override fun onBeforeBlock(handler: Runnable) {} + + override fun getCurrentHeight(): Long? { + return referenceHead.getCurrentHeight() + } + + override fun isRunning(): Boolean { + return cacheSub != null + } + + override fun start() { + cacheSub?.dispose() + sources.forEach { head -> + if (head is Lifecycle && !head.isRunning()) { + head.start() + } + } + if (referenceHead is Lifecycle && !referenceHead.isRunning()) { + referenceHead.start() + } + cacheSub = Flux.merge(sources.map { it.getFlux() }).subscribe { block -> + if (block.enriched) { + enrichedBlocks.put(block.hash, block) + enrichedPromises.get(block.hash) { Sinks.one() }.tryEmitValue(block) + } + } + } + + override fun stop() { + cacheSub?.dispose() + cacheSub = null + } + + override fun onSyncingNode(isSyncing: Boolean) {} +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt new file mode 100644 index 00000000..cbd45fea --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt @@ -0,0 +1,29 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.etherjar.domain.BlockHash +import reactor.core.publisher.Mono +import reactor.core.scheduler.Scheduler +import reactor.retry.Repeat +import java.time.Duration + +class EthereumBlockEnricher { + companion object { + fun enrich(blockHash: BlockHash, api: Reader, scheduler: Scheduler): Mono { + return Mono.just(blockHash) + .flatMap { hash -> + api.read(hash) + .subscribeOn(scheduler) + .timeout(Defaults.timeoutInternal, Mono.empty()) + }.repeatWhenEmpty { n -> + Repeat.times(5) + .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500)) + .apply(n) + } + .timeout(Defaults.timeout, Mono.empty()) + .onErrorResume { Mono.empty() } + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt index 6bd40ebb..4fc989db 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt @@ -12,6 +12,8 @@ interface EthereumLikeMultistream : Upstream, HasEgressSubscription { fun getHead(mather: Selector.Matcher): Head + fun getEnrichedHead(mather: Selector.Matcher): Head + /** * Tries to proxy the native subscribe request to the managed upstreams if * - any of them matches the matcher criteria 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 c6c0e610..2871c99b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -20,7 +20,9 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.JsonRpcReader +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.ChainFees import io.emeraldpay.dshackle.upstream.DynamicMergedHead import io.emeraldpay.dshackle.upstream.EgressSubscription @@ -38,6 +40,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream +import io.emeraldpay.etherjar.domain.BlockHash import org.springframework.cloud.sleuth.Tracer import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux @@ -209,6 +212,25 @@ open class EthereumMultistream( } } + override fun getEnrichedHead(mather: Selector.Matcher): Head = + filteredHeads.computeIfAbsent(mather.describeInternal().intern()) { _ -> + upstreams.filter { mather.matches(it) } + .apply { + log.debug("Found $size upstreams matching [${mather.describeInternal()}]") + }.let { + val selected = it.map { source -> source.getHead() } + EnrichedMergedHead( + selected, getHead(), headScheduler, + object : + Reader { + override fun read(key: BlockHash): Mono { + return reader.blocksByHashAsCont().read(key).map { res -> res.data } + } + } + ) + } + } + override fun getFeeEstimation(): ChainFees { return feeEstimation } 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 c613aa7e..81d05a7b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -16,24 +16,24 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.JsonRpcReader +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.Sinks import reactor.core.scheduler.Scheduler -import reactor.retry.Repeat import java.time.Duration class EthereumWsHead( @@ -101,7 +101,25 @@ class EthereumWsHead( block.totalDifficulty == null ) ) { - enhanceRealBlock(block) + EthereumBlockEnricher.enrich( + block.hash, + object : + Reader { + override fun read(key: BlockHash): Mono { + return api.read(JsonRpcRequest("eth_getBlockByHash", listOf(block.hash.toHex(), false))) + .flatMap { resp -> + if (resp.isNull()) { + Mono.error(SilentException("Received null for block ${block.hash}")) + } else { + Mono.just(resp) + } + } + .flatMap(JsonRpcResponse::requireResult) + .map { BlockContainer.fromEthereumJson(it, upstreamId) } + } + }, + headScheduler + ) } else { Mono.just(BlockContainer.from(block)) } @@ -113,30 +131,6 @@ class EthereumWsHead( } } - fun enhanceRealBlock(block: BlockJson): Mono { - return Mono.just(block.hash) - .flatMap { hash -> - api.read(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false))) - .flatMap { resp -> - if (resp.isNull()) { - Mono.error(SilentException("Received null for block $hash")) - } else { - Mono.just(resp) - } - } - .flatMap(JsonRpcResponse::requireResult) - .map { BlockContainer.fromEthereumJson(it, upstreamId) } - .subscribeOn(headScheduler) - .timeout(Defaults.timeoutInternal, Mono.empty()) - }.repeatWhenEmpty { n -> - Repeat.times(5) - .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500)) - .apply(n) - } - .timeout(Defaults.timeout, Mono.empty()) - .onErrorResume { Mono.empty() } - } - override fun stop() { super.stop() cancelSub() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt index fd9adec1..798bd8c0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt @@ -51,7 +51,7 @@ class ConnectBlockUpdates( fun connect() = connect(Selector.empty) override fun connect(matcher: Selector.Matcher): Flux { return connected.computeIfAbsent(matcher.describeInternal()) { key -> - extract(upstream.getHead(matcher)) + extract(upstream.getEnrichedHead(matcher)) .publishOn(scheduler) .publish() .refCount(1, Duration.ofSeconds(60)) 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 6bd8e852..cd0fb23c 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 @@ -20,7 +20,9 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.JsonRpcReader +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.ChainFees import io.emeraldpay.dshackle.upstream.DynamicMergedHead import io.emeraldpay.dshackle.upstream.EmptyHead @@ -36,6 +38,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream +import io.emeraldpay.etherjar.domain.BlockHash import org.springframework.cloud.sleuth.Tracer import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux @@ -179,6 +182,25 @@ open class EthereumPosMultiStream( } } + override fun getEnrichedHead(mather: Selector.Matcher): Head = + filteredHeads.computeIfAbsent(mather.describeInternal().intern()) { _ -> + upstreams.filter { mather.matches(it) } + .apply { + log.debug("Found $size upstreams matching [${mather.describeInternal()}]") + }.let { + val selected = it.map { source -> source.getHead() } + EnrichedMergedHead( + selected, getHead(), headScheduler, + object : + Reader { + override fun read(key: BlockHash): Mono { + return reader.blocksByHashAsCont().read(key).map { res -> res.data } + } + } + ) + } + } + override fun getFeeEstimation(): ChainFees { return feeEstimation } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 66c49072..9af3eb3e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -21,7 +21,6 @@ import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.CacheConfig -import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.EmptyReader @@ -35,9 +34,12 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import io.micrometer.core.instrument.MeterRegistry import io.micrometer.core.instrument.logging.LoggingMeterRegistry import org.apache.commons.lang3.StringUtils +import org.bouncycastle.jcajce.provider.digest.Keccak import reactor.core.scheduler.Schedulers import java.time.Instant @@ -125,13 +127,25 @@ class TestingCommons { BlockJson block = new BlockJson().tap { setNumber(height) setParentHash(BlockHash.from("0xc4b01774e426325b50f0c709753ec7cf1f1774439d587dfb91f2a4eeb8179cde")) - setHash(BlockHash.from("0xc4b01774e426325b50f0c709753ec7cf1f1774439d587dfb91f2a4eeb8179cde")) + setHash(BlockHash.from((new Keccak.Digest256()).digest(height.byteValue()))) setTotalDifficulty(BigInteger.ONE) setTimestamp(predictableTimestamp(height, 14)) } return BlockContainer.from(block) } + static BlockContainer enrichedBlockForEthereum(Long height) { + BlockJson block = new BlockJson().tap { + setNumber(height) + setParentHash(BlockHash.from("0xc4b01774e426325b50f0c709753ec7cf1f1774439d587dfb91f2a4eeb8179cde")) + setHash(BlockHash.from((new Keccak.Digest256()).digest(height.byteValue()))) + setTotalDifficulty(BigInteger.ONE) + setTimestamp(predictableTimestamp(height, 14)) + setTransactions([new TransactionRefJson(TransactionId.from("0x3b23294ade15d39261245e6a3a53c3429a015891c95885b44ded29da2d60b29c"))]) + } + return BlockContainer.from(block) + } + static BlockContainer blockForBitcoin(Long height) { def parent = BlockId.from(StringUtils.leftPad(height.toString(), 64, "0")) return new BlockContainer( diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy new file mode 100644 index 00000000..b61a6ae9 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy @@ -0,0 +1,185 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.test.ApiReaderMock +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.AbstractHead +import io.emeraldpay.dshackle.upstream.BlockValidator +import io.emeraldpay.dshackle.upstream.Lifecycle +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.domain.BlockHash +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.publisher.Sinks +import reactor.core.scheduler.Schedulers +import reactor.test.StepVerifier +import spock.lang.Specification +import java.time.Duration + +class EnrichedMergedHeadSpec extends Specification { + + def "ensures that heads are running on start"() { + setup: + def head1 = Mock(TestHead) { + _ * isRunning() >> false + _ * getFlux() >> Flux.empty() + } + def head2 = Mock(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> Flux.empty() + } + def head3 = Mock(TestHead) { + _ * isRunning() >> false + _ * getFlux() >> Flux.empty() + } + + def api = new ApiReaderMock() + when: + def merged = new EnrichedMergedHead([head1, head2], head3, Schedulers.parallel(), new BlockReader(api)) + merged.start() + + then: + 1 * head3.start() + 1 * head1.start() + merged.isRunning() + } + + def "if enriched block comes from reference head we pass it along instantly"() { + setup: + def block = TestingCommons.enrichedBlockForEthereum(100) + def api = new ApiReaderMock() + def head = Stub(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> Flux.just(block) + } + when: + def merge = new EnrichedMergedHead([], head, Schedulers.parallel(), new BlockReader(api)) + + then: + StepVerifier.create(merge.getFlux()) + .then { merge.start() } + .expectNext(block) + .thenCancel() + .verify(Duration.ofMillis(100)) + block.enriched + } + + def "enriched block arrived in sources before reference block"() { + setup: + def enrichedBlock = TestingCommons.enrichedBlockForEthereum(100) + def block = TestingCommons.blockForEthereum(100) + Sinks.Many refSink = Sinks.many().multicast().directBestEffort() + def headRef = Stub(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> refSink.asFlux() + } + def headSource = Stub(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> Flux.just(enrichedBlock) + } + when: + def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.parallel(), new BlockReader(new ApiReaderMock())) + then: + StepVerifier.create(merge.getFlux()) + .then { merge.start() } + .expectNoEvent(Duration.ofMillis(100)) + .then { refSink.tryEmitNext(block) } + .expectNext(enrichedBlock) + .thenCancel() + .verify(Duration.ofMillis(200)) + } + + def "enriched block arrived in source after reference block, but before deadline"() { + setup: + def enrichedBlock = TestingCommons.enrichedBlockForEthereum(100) + def block = TestingCommons.blockForEthereum(100) + Sinks.Many sourceSink = Sinks.many().multicast().directBestEffort() + def headRef = Stub(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> Flux.just(block) + } + def headSource = Stub(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> sourceSink.asFlux() + } + when: + def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.parallel(), new BlockReader(new ApiReaderMock())) + then: + StepVerifier.create(merge.getFlux()) + .then { merge.start() } + .expectNoEvent(Duration.ofMillis(600)) + .then { sourceSink.tryEmitNext(enrichedBlock) } + .expectNext(enrichedBlock) + .thenCancel() + .verify(Duration.ofSeconds(1)) + } + + def "enriched blocks does not arrive before deadline"() { + setup: + def enrichedBlock = TestingCommons.enrichedBlockForEthereum(100) + def block = TestingCommons.blockForEthereum(100) + def headRef = Stub(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> Flux.just(block) + } + def headSource = Stub(TestHead) { + _ * isRunning() >> true + _ * getFlux() >> Flux.just(block) + } + def api = new ApiReaderMock().tap { + answer("eth_getBlockByHash", [block.hash.toHexWithPrefix(), false], enrichedBlock.toBlock()) + } + when: + def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.parallel(), new BlockReader(api)) + then: + StepVerifier.create(merge.getFlux()) + .then { merge.start() } + .expectNoEvent(Duration.ofSeconds(1)) + .expectNext(enrichedBlock) + .thenCancel() + .verify(Duration.ofMillis(1200)) + } + + class BlockReader implements Reader { + private ApiReaderMock mockApi + BlockReader(ApiReaderMock api) { + mockApi = api + } + + Mono read(BlockHash hash) { + return mockApi.read(new JsonRpcRequest("eth_getBlockByHash", [hash.toHex(), false])) + .map { + def t = it + def a = 1 + return it + } + .flatMap(JsonRpcResponse::requireResult) + .map { BlockContainer.fromEthereumJson(it, "test") } + } + } + + class TestHead extends AbstractHead implements Lifecycle { + + TestHead() { + super(new MostWorkForkChoice(), Schedulers.parallel(), new BlockValidator.AlwaysValid(), 100_000) + } + + @Override + void start() { + + } + + @Override + void stop() { + + } + + @Override + boolean isRunning() { + return false + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy index 13ff8503..a2b85e6b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy @@ -233,7 +233,7 @@ class ConnectBlockUpdatesSpec extends Specification { 1 * getFlux() >> Flux.never() } def up = Mock(EthereumMultistream) { - 1 * getHead(Selector.empty) >> head + 1 * getEnrichedHead(Selector.empty) >> head } def connectBlockUpdates = new ConnectBlockUpdates(up, Schedulers.parallel())