diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt new file mode 100644 index 00000000..aa30d10f --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt @@ -0,0 +1,50 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import org.slf4j.LoggerFactory +import reactor.core.Disposable +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.publisher.TopicProcessor +import java.util.concurrent.atomic.AtomicReference + +open class DefaultEthereumHead: EthereumHead { + + private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java) + private val head = AtomicReference>(null) + private val stream: TopicProcessor> = TopicProcessor.create() + + fun follow(source: Flux>): Disposable { + return source.distinctUntilChanged { + it.hash + }.filter { block -> + val curr = head.get() + curr == null || curr.totalDifficulty < block.totalDifficulty + } + .subscribe { block -> + val prev = head.getAndUpdate { curr -> + if (curr == null || curr.totalDifficulty < block.totalDifficulty) { + block + } else { + curr + } + } + if (prev == null || prev.hash != block.hash) { + log.debug("New block ${block.number} ${block.hash}") + stream.onNext(block) + } + } + } + + override fun getFlux(): Flux> { + return Flux.merge( + Mono.justOrEmpty(head.get()), + Flux.from(stream) + ).onBackpressureLatest() + } + + fun getCurrent(): BlockJson? { + return head.get() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt index b5304472..ce28dca9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt @@ -42,7 +42,7 @@ open class DirectEthereumApi( } return result .doOnError { t -> - log.warn("Upstream error: [${t.message}] for ${method}") + log.warn("Upstream error: [${t.message}] for $method") } .map { val resp = ResponseJson() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt index ffe3682d..9f3f6ab3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt @@ -26,42 +26,17 @@ import reactor.core.publisher.Mono import java.util.concurrent.atomic.AtomicReference class EthereumHeadMerge( - fluxes: Iterable>> -): EthereumHead, Lifecycle { + private val fluxes: Iterable>> +): DefaultEthereumHead(), Lifecycle { - private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java) - private val flux: Flux> - private val head = AtomicReference>(null) private var subscription: Disposable? = null - init { - flux = Flux.merge(fluxes) - .distinctUntilChanged { - it.hash - } - .filter { - val curr = head.get() - curr == null || curr.totalDifficulty < it.totalDifficulty - } - .publish() - .autoConnect() - } - override fun isRunning(): Boolean { return subscription != null } override fun start() { - subscription = Flux.from(flux).subscribe { - head.set(it) - } - } - - override fun getFlux(): Flux> { - return Flux.merge( - Mono.justOrEmpty(head.get()), - Flux.from(this.flux) - ).onBackpressureLatest() + subscription = super.follow(Flux.merge(fluxes)) } override fun stop() { 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 51637ca5..6acece47 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -15,55 +15,42 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.Commands -import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.publisher.TopicProcessor import java.time.Duration -import java.util.concurrent.atomic.AtomicReference class EthereumRpcHead( - private val api: DirectEthereumApi -): EthereumHead, Lifecycle { + private val api: DirectEthereumApi, + private val interval: Duration = Duration.ofSeconds(10) +): DefaultEthereumHead(), Lifecycle { private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java) - private val head = AtomicReference>(null) - private val stream: TopicProcessor> = TopicProcessor.create() private var refreshSubscription: Disposable? = null override fun start() { - refreshSubscription = Flux.interval(Duration.ofSeconds(7)) + val base = Flux.interval(interval) .flatMap { val batch = Batch() val f = batch.add(Commands.eth().blockNumber) api.rpcClient.execute(batch) - Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5)) + Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty()) } .flatMap { val batch = Batch() val f = batch.add(Commands.eth().getBlock(it)) api.rpcClient.execute(batch) - Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5)) + Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty()) } .onErrorContinue { err, _ -> log.debug("RPC error ${err.message}") } - .distinctUntilChanged { it.hash } - .filter { block -> - val curr = head.get() - curr == null || curr.totalDifficulty < block.totalDifficulty - } - .subscribe { block -> - head.set(block) - stream.onNext(block) - } + refreshSubscription = super.follow(base) } override fun isRunning(): Boolean { @@ -75,11 +62,4 @@ class EthereumRpcHead( refreshSubscription = null } - override fun getFlux(): Flux> { - return Flux.merge( - Mono.justOrEmpty(head.get()), - Flux.from(stream) - ).onBackpressureLatest() - } - } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index 07453196..6e2c5d64 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -21,6 +21,7 @@ import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable +import java.time.Duration open class EthereumUpstream( private val id: String, @@ -70,21 +71,20 @@ open class EthereumUpstream( override fun stop() { validatorSubscription?.dispose() validatorSubscription = null + if (head is Lifecycle) { + head.stop() + } } open fun createHead(): EthereumHead { return if (ethereumWs != null) { - // load current block through RPC then listen for following blocks through WS val ws = EthereumWsHead(ethereumWs).apply { this.start() } - val rpc = EthereumRpcHead(api).apply { + val rpc = EthereumRpcHead(api, Duration.ofSeconds(20)).apply { this.start() } - val currentHead = rpc.getFlux().next().doFinally { - rpc.stop() - } - EthereumHeadMerge(listOf(currentHead, ws.getFlux())).apply { + EthereumHeadMerge(listOf(rpc.getFlux(), ws.getFlux())).apply { this.start() } } else { @@ -106,10 +106,6 @@ open class EthereumUpstream( return api } - fun getApi(): DirectEthereumApi { - return api - } - override fun getOptions(): UpstreamsConfig.Options { return options } 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 ab3cd68f..29f6c0d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -15,48 +15,24 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import java.util.concurrent.atomic.AtomicReference class EthereumWsHead( private val ws: EthereumWs -): EthereumHead, Lifecycle { +): DefaultEthereumHead(), Lifecycle { private val log = LoggerFactory.getLogger(EthereumWsHead::class.java) private var subscription: Disposable? = null - private val head = AtomicReference>(null) - private var stream: Flux>? = null - - override fun getFlux(): Flux> { - return stream?.let { - Flux.merge( - Mono.justOrEmpty(head.get()), - Flux.from(this.stream) - ).onBackpressureLatest() - } ?: Flux.error(Exception("Not started")) - } override fun isRunning(): Boolean { return subscription != null } override fun start() { - val flux = ws.getFlux() - .distinctUntilChanged { it.hash } - .filter { block -> - val curr = head.get() - curr == null || curr.totalDifficulty < block.totalDifficulty - }.share() - - this.subscription = flux.subscribe(head::set) - this.stream = flux + this.subscription = super.follow(ws.getFlux()) } override fun stop() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt index f3657007..ab29b001 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -22,6 +22,7 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.* +import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.emeraldpay.grpc.Chain @@ -35,7 +36,6 @@ import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toMono import java.lang.Exception import java.math.BigInteger @@ -57,10 +57,8 @@ open class GrpcUpstream( private val log = LoggerFactory.getLogger(GrpcUpstream::class.java) private val options = UpstreamsConfig.Options.getDefaults() - private val headBlock = AtomicReference>(null) - private val streamBlocks: TopicProcessor> = TopicProcessor.create() private val nodes = AtomicReference(NodeDetailsList()) - private val head = Head(this) + private val head = DefaultEthereumHead() private var targets: CallMethods? = null private var headSubscription: Disposable? = null @@ -110,40 +108,36 @@ open class GrpcUpstream( internal fun observeHead(flux: Flux) { - headSubscription = flux.map { value -> - val block = BlockJson() - block.number = value.height - block.totalDifficulty = BigInteger(1, value.weight.toByteArray()) - block.hash = BlockHash.from("0x"+value.blockId) - block - } - .distinctUntilChanged { it.hash } - .filter { block -> - val curr = headBlock.get() - curr == null || curr.totalDifficulty < block.totalDifficulty - } - .flatMap { - getApi(Selector.EmptyMatcher()) - .executeAndConvert(Commands.eth().getBlock(it.hash)) - .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream"))) - .doOnError { t -> - val msg = "Failed to download block data for chain $chain" - if (t is RpcException) { - log.warn("$msg. Message: ${t.message}") - } else { - log.error(msg, t) - } - } - } - .onErrorContinue { err, _ -> - log.error("Head subscription error: ${err.message}") - } - .subscribe { block -> - log.debug("New block ${block.number} on ${chain}") - setStatus(UpstreamAvailability.OK) - headBlock.set(block) - streamBlocks.onNext(block) - } + val base = flux.map { value -> + val block = BlockJson() + block.number = value.height + block.totalDifficulty = BigInteger(1, value.weight.toByteArray()) + block.hash = BlockHash.from("0x"+value.blockId) + block + }.distinctUntilChanged { + it.hash + }.filter { block -> + val curr = head.getCurrent() + curr == null || curr.totalDifficulty < block.totalDifficulty + }.flatMap { + getApi(Selector.EmptyMatcher()) + .executeAndConvert(Commands.eth().getBlock(it.hash)) + .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream"))) + .doOnError { t -> + val msg = "Failed to download block data for chain $chain" + if (t is RpcException) { + log.warn("$msg. Message: ${t.message}") + } else { + log.error(msg, t) + } + } + }.onErrorContinue { err, _ -> + log.error("Head subscription error: ${err.message}") + }.doOnNext { + setStatus(UpstreamAvailability.OK) + } + + headSubscription = head.follow(base) } fun init(conf: BlockchainOuterClass.DescribeChain) { @@ -191,7 +185,7 @@ open class GrpcUpstream( } override fun isAvailable(): Boolean { - return getStatus() == UpstreamAvailability.OK && headBlock.get() != null && nodes.get().getNodes().any { + return getStatus() == UpstreamAvailability.OK && head.getCurrent() != null && nodes.get().getNodes().any { it.quorum > 0 } } @@ -208,16 +202,4 @@ open class GrpcUpstream( return options } - class Head( - val upstream: GrpcUpstream - ): EthereumHead { - - override fun getFlux(): Flux> { - return Flux.merge( - Mono.justOrEmpty(upstream.headBlock.get()), - Flux.from(upstream.streamBlocks) - ) - } - } - } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy new file mode 100644 index 00000000..6e642416 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy @@ -0,0 +1,114 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Flux +import reactor.test.StepVerifier +import spock.lang.Specification + +class DefaultEthereumHeadSpec extends Specification { + + DefaultEthereumHead head = new DefaultEthereumHead() + + def blocks = (10L..20L).collect { i -> + new BlockJson().with { + it.number = 10000L + i + it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec89152" + i) + it.totalDifficulty = 11 * i + return it + } + } + + def "Starts to follow"() { + when: + head.follow(Flux.just(blocks[0])) + def act = head.flux + then: + StepVerifier.create(act) + .expectNext(blocks[0]) + .expectComplete() + } + + def "Follows normal order"() { + when: + head.follow(Flux.just(blocks[0], blocks[1], blocks[3])) + def act = head.flux + then: + StepVerifier.create(act) + .expectNext(blocks[0]) + .expectNext(blocks[1]) + .expectNext(blocks[3]) + .expectComplete() + } + + def "Ignores old"() { + when: + head.follow(Flux.just(blocks[0], blocks[3], blocks[1])) + def act = head.flux + then: + StepVerifier.create(act) + .expectNext(blocks[0]) + .expectNext(blocks[3]) + .expectComplete() + } + + def "Ignores repeating"() { + when: + head.follow(Flux.just(blocks[0], blocks[3], blocks[3], blocks[3], blocks[2], blocks[3])) + def act = head.flux + then: + StepVerifier.create(act) + .expectNext(blocks[0]) + .expectNext(blocks[3]) + .expectComplete() + } + + def "Ignores less difficult"() { + when: + def block3less = new BlockJson().with { + it.number = blocks[3].number + it.hash = blocks[3].hash + it.totalDifficulty = blocks[3].totalDifficulty - 1 + return it + } + head.follow(Flux.just(blocks[0], blocks[3], block3less)) + def act = head.flux + then: + StepVerifier.create(act) + .expectNext(blocks[0]) + .expectNext(blocks[3]) + .expectComplete() + } + + def "Replaces with more difficult"() { + when: + def block3less = new BlockJson().with { + it.number = blocks[3].number + it.hash = blocks[3].hash + it.totalDifficulty = blocks[3].totalDifficulty + 1 + return it + } + head.follow(Flux.just(blocks[0], blocks[3], block3less)) + def act = head.flux + then: + StepVerifier.create(act) + .expectNext(blocks[0]) + .expectNext(block3less) + .expectComplete() + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy index 5bfc4be6..c7f501ca 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy @@ -32,6 +32,7 @@ import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.emerald.EmeraldGrpcTransport import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration @@ -90,9 +91,6 @@ class GrpcUpstreamSpec extends Specification { def "Follows difficulty, ignores less difficult"() { setup: - def callData = [:] - def finished = new CompletableFuture() - def chain = Chain.ETHEREUM def api = TestingCommons.api(Stub(RpcClient)) def block1 = new BlockJson().with { it.number = 650246 @@ -130,19 +128,17 @@ class GrpcUpstreamSpec extends Specification { .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray())) .build() ) - finished.complete(true) } }) def transport = EmeraldGrpcTransport.newBuilder().forChannel(client.channel).build() - def upstream = new GrpcUpstream("test", chain, client, objectMapper, transport) + def upstream = new GrpcUpstream("test", Chain.ETHEREUM, client, objectMapper, transport) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods(["eth_getBlockByHash"]) .build()) when: upstream.start() - finished.get() - def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1)) + def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block() then: upstream.status == UpstreamAvailability.OK h.hash == BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -203,7 +199,7 @@ class GrpcUpstreamSpec extends Specification { when: upstream.start() finished.get() - def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1)) + def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block() then: upstream.status == UpstreamAvailability.OK h.hash == BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")