Enriched merge head for ethereum subscription (#262)

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(

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<BlockContainer> 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<BlockContainer> 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<BlockHash, BlockContainer> {
+        private ApiReaderMock mockApi
+        BlockReader(ApiReaderMock api) {
+            mockApi = api
+        }
+
+        Mono<BlockContainer> 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
+        }
+    }
+}

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