diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt index 9f8a1911..51af4aee 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt @@ -21,8 +21,10 @@ 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.util.function.Tuple2 import reactor.util.function.Tuples +import java.time.Duration /** * Observer group of upstreams and defined a distance in blocks (lag) between a leader (best height/difficulty) and @@ -38,6 +40,7 @@ abstract class HeadLagObserver( private var current: Disposable? = null override fun start() { + current?.dispose() current = subscription().subscribe { } } @@ -61,12 +64,15 @@ abstract class HeadLagObserver( fun probeFollowers(top: BlockContainer): Flux> { return Flux.fromIterable(followers) .parallel(followers.size) - .flatMap { mapLagging(top, it, getCurrentBlocks(it)) } + .flatMap { up -> mapLagging(top, up, getCurrentBlocks(up)).subscribeOn(Schedulers.elastic()) } .sequential() .onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) } } - abstract fun getCurrentBlocks(up: Upstream): Flux + open fun getCurrentBlocks(up: Upstream): Flux { + val head = up.getHead() + return head.getFlux().take(Duration.ofSeconds(1)) + } fun mapLagging(top: BlockContainer, up: Upstream, blocks: Flux): Flux> { return blocks @@ -78,11 +84,14 @@ abstract class HeadLagObserver( } } - abstract fun extractDistance(top: BlockContainer, curr: BlockContainer): Long - - fun forkDistance(top: BlockContainer, curr: BlockContainer): Long { - //TODO look for common ancestor? though it may be a corruption - return 6 + open fun extractDistance(top: BlockContainer, curr: BlockContainer): Long { + return when { + curr.height > top.height -> if (curr.difficulty >= top.difficulty) 0 else forkDistance(top, curr) + curr.height == top.height -> if (curr.difficulty == top.difficulty) 0 else forkDistance(top, curr) + else -> top.height - curr.height + } } + abstract fun forkDistance(top: BlockContainer, curr: BlockContainer): Long + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt new file mode 100644 index 00000000..28b3ad94 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt @@ -0,0 +1,38 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * + * 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.bitcoin + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.HeadLagObserver +import io.emeraldpay.dshackle.upstream.Upstream +import org.slf4j.LoggerFactory + +class BitcoinHeadLagObserver( + master: Head, + followers: Collection +) : HeadLagObserver(master, followers) { + + companion object { + private val log = LoggerFactory.getLogger(BitcoinHeadLagObserver::class.java) + } + + override fun forkDistance(top: BlockContainer, curr: BlockContainer): Long { + //TODO fetch actual blocks + return 3 + } + +} \ No newline at end of file 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 9de37b73..8f647089 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -64,8 +64,8 @@ open class BitcoinMultistream( val newHead = MergedHead(upstreams.map { it.getHead() }).apply { this.start() } -// val lagObserver = TODO -// this.lagObserver = lagObserver + val lagObserver = BitcoinHeadLagObserver(newHead, upstreams) + this.lagObserver = lagObserver newHead } onHeadUpdated(head) 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 1ae739d9..59ac966b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt @@ -33,16 +33,9 @@ class EthereumHeadLagObserver( private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java) } - override fun getCurrentBlocks(up: Upstream): Flux { - val head = up.getHead() - return head.getFlux().take(Duration.ofSeconds(1)) + override fun forkDistance(top: BlockContainer, curr: BlockContainer): Long { + //TODO look for common ancestor? though it may be a corruption + return 6 } - override fun extractDistance(top: BlockContainer, curr: BlockContainer): Long { - return when { - curr.height > top.height -> if (curr.difficulty >= top.difficulty) 0 else forkDistance(top, curr) - curr.height == top.height -> if (curr.difficulty == top.difficulty) 0 else forkDistance(top, curr) - else -> top.height - curr.height - } - } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserverSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy similarity index 85% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserverSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy index 0d255b73..4ccf5868 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserverSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.emeraldpay.dshackle.upstream.ethereum +package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.Head @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.HeadLagObserver import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.json.BlockJson +import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux import reactor.core.publisher.TopicProcessor import reactor.test.StepVerifier @@ -31,7 +32,7 @@ import spock.lang.Specification import java.time.Duration import java.time.Instant -class EthereumHeadLagObserverSpec extends Specification { +class HeadLagObserverSpec extends Specification { def "Updates lag distance"() { setup: @@ -72,7 +73,7 @@ class EthereumHeadLagObserverSpec extends Specification { 1 * up2.setLag(1) 1 * up2.setLag(0) - HeadLagObserver observer = new EthereumHeadLagObserver(master, [up1, up2]) + HeadLagObserver observer = new TestHeadLagObserver(master, [up1, up2]) when: def act = observer.subscription().take(Duration.ofMillis(1200)) @@ -86,7 +87,7 @@ class EthereumHeadLagObserverSpec extends Specification { def "Probes until there is no difference"() { setup: Head master = Mock() - HeadLagObserver observer = new EthereumHeadLagObserver(master, []) + HeadLagObserver observer = new TestHeadLagObserver(master, []) Upstream up = Mock() def blocks = [100, 101, 102].collect { i -> @@ -113,7 +114,7 @@ class EthereumHeadLagObserverSpec extends Specification { def "Correct distance"() { setup: Head master = Mock() - HeadLagObserver observer = new EthereumHeadLagObserver(master, []) + HeadLagObserver observer = new TestHeadLagObserver(master, []) expect: def top = new BlockJson().with { it.number = topHeight @@ -140,8 +141,20 @@ class EthereumHeadLagObserverSpec extends Specification { 100 | 1000 | 101 | 1010 | 0 100 | 1000 | 102 | 1020 | 0 - 100 | 1000 | 100 | 1010 | 6 - 100 | 1100 | 100 | 1000 | 6 + 100 | 1000 | 100 | 1010 | 11 + 100 | 1100 | 100 | 1000 | 11 } + + class TestHeadLagObserver extends HeadLagObserver { + + TestHeadLagObserver(@NotNull Head master, @NotNull Collection followers) { + super(master, followers) + } + + @Override + long forkDistance(@NotNull BlockContainer top, @NotNull BlockContainer curr) { + return 11 + } + } }