From 55e2dc70c4fc379448b59aa7ea832a85998ad140 Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Thu, 11 Apr 2024 18:03:32 +0400 Subject: [PATCH] Refactoring lower bounds (#450) --- .../io/emeraldpay/dshackle/rpc/Describe.kt | 2 - .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 45 +++++++- .../configure/GenericUpstreamCreator.kt | 2 +- .../upstream/LowerBoundBlockDetector.kt | 87 --------------- .../dshackle/upstream/Multistream.kt | 24 ++-- .../RecursiveLowerBoundBlockDetector.kt | 79 -------------- .../emeraldpay/dshackle/upstream/Upstream.kt | 3 +- .../BeaconChainLowerBoundBlockDetector.kt | 21 ---- .../BeaconChainLowerBoundService.kt | 15 +++ .../BeaconChainLowerBoundStateDetector.kt | 17 +++ .../beaconchain/BeaconChainSpecific.kt | 6 +- .../upstream/bitcoin/BitcoinRpcUpstream.kt | 6 +- .../ethereum/EthereumChainSpecific.kt | 6 +- .../ethereum/EthereumLowerBoundService.kt | 15 +++ ....kt => EthereumLowerBoundStateDetector.kt} | 44 ++++---- .../upstream/generic/ChainSpecific.kt | 4 +- .../upstream/generic/GenericUpstream.kt | 14 +-- .../upstream/grpc/BitcoinGrpcUpstream.kt | 6 +- .../upstream/grpc/GenericGrpcUpstream.kt | 6 +- .../upstream/lowerbound/LowerBoundData.kt | 23 ++++ .../upstream/lowerbound/LowerBoundDetector.kt | 45 ++++++++ .../upstream/lowerbound/LowerBoundService.kt | 32 ++++++ .../detector/RecursiveLowerBound.kt | 103 ++++++++++++++++++ .../upstream/near/NearChainSpecific.kt | 6 +- .../near/NearLowerBoundBlockDetector.kt | 26 ----- .../upstream/near/NearLowerBoundService.kt | 15 +++ .../near/NearLowerBoundStateDetector.kt | 27 +++++ .../polkadot/PolkadotChainSpecific.kt | 6 +- .../PolkadotLowerBoundBlockDetector.kt | 47 -------- .../polkadot/PolkadotLowerBoundService.kt | 16 +++ .../PolkadotLowerBoundStateDetector.kt | 51 +++++++++ .../upstream/solana/SolanaChainSpecific.kt | 6 +- .../solana/SolanaLowerBoundService.kt | 15 +++ ...tor.kt => SolanaLowerBoundSlotDetector.kt} | 30 ++--- .../starknet/StarknetChainSpecific.kt | 6 +- .../StarknetLowerBoundBlockDetector.kt | 21 ---- .../starknet/StarknetLowerBoundService.kt | 15 +++ .../StarknetLowerBoundStateDetector.kt | 17 +++ .../dshackle/test/GenericUpstreamMock.groovy | 13 ++- .../dshackle/upstream/FilteredApisSpec.groovy | 2 +- ...t.kt => RecursiveLowerBoundServiceTest.kt} | 44 +++++--- ...Test.kt => SolanaLowerBoundServiceTest.kt} | 23 ++-- .../StarknetLowerBoundBlockDetectorTest.kt | 24 ---- .../StarknetLowerBoundStateDetectorTest.kt | 22 ++++ 44 files changed, 619 insertions(+), 418 deletions(-) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumLowerBoundBlockDetector.kt => EthereumLowerBoundStateDetector.kt} (59%) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundService.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundBlockDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundService.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundStateDetector.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundService.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundStateDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundService.kt rename src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/{SolanaLowerBoundBlockDetector.kt => SolanaLowerBoundSlotDetector.kt} (77%) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundService.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetector.kt rename src/test/kotlin/io/emeraldpay/dshackle/upstream/{RecursiveLowerBoundBlockDetectorTest.kt => RecursiveLowerBoundServiceTest.kt} (77%) rename src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/{SolanaLowerBoundBlockDetectorTest.kt => SolanaLowerBoundServiceTest.kt} (69%) delete mode 100644 src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetectorTest.kt create mode 100644 src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetectorTest.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 354c05ff..35075554 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -46,8 +46,6 @@ class Describe( .addAllSupportedSubscriptions(chainUpstreams.getEgressSubscription().getAvailableTopics()) .setStatus(status) .setCurrentHeight(chainUpstreams.getHead().getCurrentHeight() ?: 0) - .setCurrentLowerBlock(chainUpstreams.getLowerBlock().blockNumber) - .setCurrentLowerSlot(chainUpstreams.getLowerBlock().slot ?: 0) chainUpstreams.getQuorumLabels() .forEach { node -> val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 792bbb2c..a3af821d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -23,6 +23,8 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -51,17 +53,54 @@ class StreamHead( } fun asProto(ms: Multistream, chain: Chain, block: BlockContainer): BlockchainOuterClass.ChainHead { + val msLowerBounds = ms.getLowerBounds() + val lowerBoundsProto = msLowerBounds + .map { + BlockchainOuterClass.LowerBound.newBuilder() + .setLowerBoundTimestamp(it.timestamp) + .setLowerBoundType(toProtoLowerBoundType(it.type)) + .setLowerBoundValue(it.lowerBound) + .build() + } + val toOldApi = toOldApi(msLowerBounds) + return BlockchainOuterClass.ChainHead.newBuilder() .setChainValue(chain.id) .setHeight(block.height) .setSlot(block.slot) - .setCurrentLowerBlock(ms.getLowerBlock().blockNumber) - .setCurrentLowerSlot(ms.getLowerBlock().slot ?: 0) - .setCurrentLowerDataTimestamp(ms.getLowerBlock().timestamp) + .setCurrentLowerBlock(toOldApi.block) + .setCurrentLowerSlot(toOldApi.slot) + .setCurrentLowerDataTimestamp(toOldApi.timestamp) + .addAllLowerBounds(lowerBoundsProto) .setTimestamp(block.timestamp.toEpochMilli()) .setWeight(ByteString.copyFrom(block.difficulty.toByteArray())) .setBlockId(block.hash.toHex()) .setParentBlockId(block.parentHash?.toHex() ?: "") .build() } + + private fun toOldApi(lowerBounds: Collection): LowerBoundDataOldApiCompatibility { + val lowerBlockData = lowerBounds.find { it.type == LowerBoundType.STATE } ?: LowerBoundData.default() + val slot = lowerBounds.find { it.type == LowerBoundType.SLOT }?.lowerBound ?: 0 + + return LowerBoundDataOldApiCompatibility( + lowerBlockData.lowerBound, + slot, + lowerBlockData.timestamp, + ) + } + + private fun toProtoLowerBoundType(type: LowerBoundType): BlockchainOuterClass.LowerBoundType { + return when (type) { + LowerBoundType.SLOT -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_SLOT + LowerBoundType.UNKNOWN -> BlockchainOuterClass.LowerBoundType.UNRECOGNIZED + LowerBoundType.STATE -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_STATE + } + } + + private data class LowerBoundDataOldApiCompatibility( + val block: Long, + val slot: Long, + val timestamp: Long, + ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt index 7e08a7df..f3c4e9fb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt @@ -86,7 +86,7 @@ open class GenericUpstreamCreator( connectorFactory, cs::validator, cs::upstreamSettingsDetector, - cs::lowerBoundBlockDetector, + cs::lowerBoundService, ) upstream.start() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt deleted file mode 100644 index aeb9dae2..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt +++ /dev/null @@ -1,87 +0,0 @@ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.Chain -import org.slf4j.LoggerFactory -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.kotlin.core.publisher.switchIfEmpty -import java.time.Duration -import java.time.Instant -import java.util.concurrent.atomic.AtomicBoolean -import java.util.concurrent.atomic.AtomicReference - -typealias LowerBoundBlockDetectorBuilder = (Chain, Upstream) -> LowerBoundBlockDetector - -fun Long.toHex() = "0x${this.toString(16)}" - -abstract class LowerBoundBlockDetector( - private val chain: Chain, - private val upstream: Upstream, -) { - private val currentLowerBlock = AtomicReference(LowerBlockData.default()) - - protected val log = LoggerFactory.getLogger(this::class.java) - - fun lowerBlock(): Flux { - val notProcessing = AtomicBoolean(true) - - return Flux.interval( - Duration.ofSeconds(15), - Duration.ofMinutes(periodRequest()), - ) - .filter { notProcessing.get() } - .flatMap { - notProcessing.set(false) - lowerBlockDetect() - .onErrorResume { Mono.just(LowerBlockData.default()) } - .switchIfEmpty { Mono.just(LowerBlockData.default()) } // just to trigger onNext event - } - .doOnNext { - notProcessing.set(true) - } - .filter { it.blockNumber > currentLowerBlock.get().blockNumber } - .map { - log.info("Lower block of ${upstream.getId()} $chain: block height - {}, slot - {}", it.blockNumber, it.slot ?: "NA") - - currentLowerBlock.set(it) - it - } - } - - fun getCurrentLowerBlock(): LowerBlockData = currentLowerBlock.get() - - protected abstract fun lowerBlockDetect(): Mono - - protected abstract fun periodRequest(): Long - - data class LowerBlockData( - val blockNumber: Long, - val slot: Long?, - val timestamp: Long, - ) : Comparable { - constructor(blockNumber: Long) : this(blockNumber, null, Instant.now().epochSecond) - - constructor(blockNumber: Long, slot: Long) : this(blockNumber, slot, Instant.now().epochSecond) - - companion object { - fun default() = LowerBlockData(0, 0, 0) - } - - override fun compareTo(other: LowerBlockData): Int { - return this.blockNumber.compareTo(other.blockNumber) - } - } - - data class LowerBoundData( - val left: Long, - val right: Long, - val current: Long, - val found: Boolean, - ) { - constructor(left: Long, right: Long) : this(left, right, 0, false) - - constructor(left: Long, right: Long, current: Long) : this(left, right, current, false) - - constructor(current: Long, found: Boolean) : this(0, 0, current, found) - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index dbee254b..642a4eea 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -28,6 +28,8 @@ import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallSelector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import io.micrometer.core.instrument.Gauge import io.micrometer.core.instrument.Meter import io.micrometer.core.instrument.Metrics @@ -78,8 +80,7 @@ abstract class Multistream( @Volatile private var capabilities: Set = emptySet() - @Volatile - private var lowerBlock: LowerBoundBlockDetector.LowerBlockData = LowerBoundBlockDetector.LowerBlockData.default() + private val lowerBounds = ConcurrentHashMap() @Volatile private var quorumLabels: List? = null @@ -242,10 +243,15 @@ abstract class Multistream( } } quorumLabels = getQuorumLabels(availableUpstreams) + availableUpstreams - .filter { it.getLowerBlock() != LowerBoundBlockDetector.LowerBlockData.default() } - .minOfOrNull { it.getLowerBlock() } - ?.let { lowerBlock = it } + .flatMap { it.getLowerBounds() } + .groupBy { it.type } + .forEach { entry -> + val min = entry.value.minBy { it.lowerBound } + lowerBounds[entry.key] = min + } + when { upstreams.size == 1 -> { lagObserver?.stop() @@ -332,7 +338,9 @@ abstract class Multistream( started = true } - override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData = lowerBlock + override fun getLowerBounds(): Collection { + return lowerBounds.values + } override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { return Upstream.UpstreamSettingsData( @@ -434,10 +442,10 @@ abstract class Multistream( val weak = getUpstreams() .filter { it.getStatus() != UpstreamAvailability.OK } .joinToString(", ") { it.getId() } - val lowerBlockData = "[height=${lowerBlock.blockNumber}, slot=${lowerBlock.slot ?: "NA"}]" + val lowerBlockData = lowerBounds.entries.joinToString(", ") { "${it.key}=${it.value.lowerBound}" } val instance = System.identityHashCode(this).toString(16) - log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], lower block=$lowerBlockData, weak=[$weak] ($instance)") + log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], lower bounds=[$lowerBlockData], weak=[$weak] ($instance)") } fun test(event: UpstreamChangeEvent): Boolean { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt deleted file mode 100644 index fa319051..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt +++ /dev/null @@ -1,79 +0,0 @@ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.Chain -import reactor.core.publisher.Mono -import reactor.util.retry.Retry -import reactor.util.retry.RetryBackoffSpec -import java.time.Duration - -abstract class RecursiveLowerBoundBlockDetector( - chain: Chain, - private val upstream: Upstream, -) : LowerBoundBlockDetector(chain, upstream) { - - override fun lowerBlockDetect(): Mono { - return Mono.just(upstream.getHead()) - .flatMap { - val currentHeight = it.getCurrentHeight() - if (currentHeight == null) { - Mono.empty() - } else { - Mono.just(LowerBoundData(0, currentHeight)) - } - } - .expand { data -> - if (data.found) { - Mono.empty() - } else { - val middle = middleBlock(data) - - if (data.left > data.right) { - val current = if (data.current == 0L) 1 else data.current - Mono.just(LowerBoundData(current, true)) - } else { - hasState(middle) - .map { - if (it) { - LowerBoundData(data.left, middle - 1, middle) - } else { - LowerBoundData(middle + 1, data.right, data.current) - } - } - } - } - } - .filter { it.found } - .next() - .map { - LowerBlockData(it.current) - } - } - - private fun middleBlock(lowerBoundData: LowerBoundData): Long = - lowerBoundData.left + (lowerBoundData.right - lowerBoundData.left) / 2 - - protected fun retrySpec(nonRetryableErrors: Set): RetryBackoffSpec { - return Retry.backoff( - Long.MAX_VALUE, - Duration.ofSeconds(1), - ) - .maxBackoff(Duration.ofMinutes(3)) - .filter { - !nonRetryableErrors.any { err -> it.message?.contains(err, true) ?: false } - } - .doAfterRetry { - log.debug( - "Error in calculation of lower block of upstream {}, retry attempt - {}, message - {}", - upstream.getId(), - it.totalRetries(), - it.failure().message, - ) - } - } - - override fun periodRequest(): Long { - return 10 - } - - protected abstract fun hasState(blockNumber: Long): Mono -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index f2625f38..986db4d0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import reactor.core.publisher.Flux interface Upstream : Lifecycle { @@ -44,7 +45,7 @@ interface Upstream : Lifecycle { fun getId(): String fun getCapabilities(): Set fun isGrpc(): Boolean - fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData + fun getLowerBounds(): Collection fun getUpstreamSettingsData(): UpstreamSettingsData? fun cast(selfType: Class): T diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt deleted file mode 100644 index 12d62e76..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt +++ /dev/null @@ -1,21 +0,0 @@ -package io.emeraldpay.dshackle.upstream.beaconchain - -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector -import io.emeraldpay.dshackle.upstream.Upstream -import reactor.core.publisher.Mono - -class BeaconChainLowerBoundBlockDetector( - chain: Chain, - upstream: Upstream, -) : LowerBoundBlockDetector(chain, upstream) { - - // TODO: consensus nodes could be launched either in full mode or in archive mode - override fun lowerBlockDetect(): Mono { - return Mono.just(LowerBlockData(1)) - } - - override fun periodRequest(): Long { - return 120 - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt new file mode 100644 index 00000000..910542b4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt @@ -0,0 +1,15 @@ +package io.emeraldpay.dshackle.upstream.beaconchain + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService + +class BeaconChainLowerBoundService( + chain: Chain, + upstream: Upstream, +) : LowerBoundService(chain, upstream) { + override fun detectors(): List { + return listOf(BeaconChainLowerBoundStateDetector()) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt new file mode 100644 index 00000000..2793c1b5 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt @@ -0,0 +1,17 @@ +package io.emeraldpay.dshackle.upstream.beaconchain + +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import reactor.core.publisher.Flux + +class BeaconChainLowerBoundStateDetector : LowerBoundDetector() { + + override fun period(): Long { + return 120 + } + + override fun internalDetectLowerBound(): Flux { + return Flux.just(LowerBoundData(1, LowerBoundType.STATE)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt index d579dd4c..338a1e5c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt @@ -12,11 +12,11 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.RestParams import java.math.BigInteger import java.time.Instant @@ -68,8 +68,8 @@ object BeaconChainSpecific : AbstractPollChainSpecific() { return BeaconChainValidator(upstream, options) } - override fun lowerBoundBlockDetector(chain: Chain, upstream: Upstream): LowerBoundBlockDetector { - return BeaconChainLowerBoundBlockDetector(chain, upstream) + override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { + return BeaconChainLowerBoundService(chain, upstream) } } 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 b925829a..9c1e1819 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt @@ -25,10 +25,10 @@ import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.HttpReader import io.emeraldpay.dshackle.upstream.Lifecycle -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import reactor.core.Disposable open class BitcoinRpcUpstream( @@ -72,8 +72,8 @@ open class BitcoinRpcUpstream( return false } - override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData { - return LowerBoundBlockDetector.LowerBlockData.default() + override fun getLowerBounds(): Collection { + return emptyList() } override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt index ff323fbf..a2edee9f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt @@ -12,7 +12,6 @@ import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription import io.emeraldpay.dshackle.upstream.LogsOracle -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector @@ -27,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder import io.emeraldpay.dshackle.upstream.generic.GenericUpstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Mono @@ -91,8 +91,8 @@ object EthereumChainSpecific : AbstractPollChainSpecific() { return EthereumUpstreamValidator(chain, upstream, options, config) } - override fun lowerBoundBlockDetector(chain: Chain, upstream: Upstream): LowerBoundBlockDetector { - return EthereumLowerBoundBlockDetector(chain, upstream) + override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { + return EthereumLowerBoundService(chain, upstream) } override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt new file mode 100644 index 00000000..eb01605e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt @@ -0,0 +1,15 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService + +class EthereumLowerBoundService( + chain: Chain, + private val upstream: Upstream, +) : LowerBoundService(chain, upstream) { + override fun detectors(): List { + return listOf(EthereumLowerBoundStateDetector(upstream)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt similarity index 59% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt index 7889c0fa..178bec3c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt @@ -1,18 +1,21 @@ package io.emeraldpay.dshackle.upstream.ethereum -import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse -import io.emeraldpay.dshackle.upstream.RecursiveLowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound +import io.emeraldpay.dshackle.upstream.lowerbound.toHex import io.emeraldpay.dshackle.upstream.rpcclient.ListParams -import io.emeraldpay.dshackle.upstream.toHex +import reactor.core.publisher.Flux import reactor.core.publisher.Mono -class EthereumLowerBoundBlockDetector( - chain: Chain, +class EthereumLowerBoundStateDetector( private val upstream: Upstream, -) : RecursiveLowerBoundBlockDetector(chain, upstream) { +) : LowerBoundDetector() { + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, nonRetryableErrors) companion object { private val nonRetryableErrors = setOf( @@ -41,19 +44,22 @@ class EthereumLowerBoundBlockDetector( ) } - override fun hasState(blockNumber: Long): Mono { - if (blockNumber == 0L) { - return Mono.just(true) + override fun period(): Long { + return 5 + } + + override fun internalDetectLowerBound(): Flux { + return recursiveLowerBound.recursiveDetectLowerBound { block -> + if (block == 0L) { + Mono.just(ChainResponse(ByteArray(0), null)) + } else { + upstream.getIngressReader().read( + ChainRequest( + "eth_getBalance", + ListParams(ZERO_ADDRESS, block.toHex()), + ), + ) + } } - return upstream.getIngressReader().read( - ChainRequest( - "eth_getBalance", - ListParams(ZERO_ADDRESS, blockNumber.toHex()), - ), - ) - .retryWhen(retrySpec(nonRetryableErrors)) - .flatMap(ChainResponse::requireResult) - .map { true } - .onErrorReturn(false) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt index 6a80732e..71428307 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -20,7 +20,6 @@ import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription import io.emeraldpay.dshackle.upstream.LogsOracle -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector @@ -30,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.near.NearChainSpecific import io.emeraldpay.dshackle.upstream.polkadot.PolkadotChainSpecific import io.emeraldpay.dshackle.upstream.solana.SolanaChainSpecific @@ -66,7 +66,7 @@ interface ChainSpecific { fun callSelector(caches: Caches): CallSelector? - fun lowerBoundBlockDetector(chain: Chain, upstream: Upstream): LowerBoundBlockDetector + fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService } object ChainSpecificRegistry { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt index 37e64908..716fc18d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt @@ -13,8 +13,6 @@ import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetectorBuilder import io.emeraldpay.dshackle.upstream.UNKNOWN_CLIENT_VERSION import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -25,6 +23,8 @@ import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundServiceBuilder import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux @@ -45,7 +45,7 @@ open class GenericUpstream( connectorFactory: ConnectorFactory, validatorBuilder: UpstreamValidatorBuilder, upstreamSettingsDetectorBuilder: UpstreamSettingsDetectorBuilder, - lowerBoundBlockDetectorBuilder: LowerBoundBlockDetectorBuilder, + lowerBoundServiceBuilder: LowerBoundServiceBuilder, ) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig), Lifecycle { private val validator: UpstreamValidator? = validatorBuilder(chain, this, getOptions(), chainConfig) @@ -58,7 +58,7 @@ open class GenericUpstream( private var livenessSubscription: Disposable? = null private val settingsDetector = upstreamSettingsDetectorBuilder(chain, this) - private val lowerBoundBlockDetector = lowerBoundBlockDetectorBuilder(chain, this) + private val lowerBoundService = lowerBoundServiceBuilder(chain, this) private val started = AtomicBoolean(false) private val isUpstreamValid = AtomicBoolean(false) @@ -90,8 +90,8 @@ open class GenericUpstream( return false } - override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData { - return lowerBoundBlockDetector.getCurrentLowerBlock() + override fun getLowerBounds(): Collection { + return lowerBoundService.getLowerBounds() } override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { @@ -233,7 +233,7 @@ open class GenericUpstream( } private fun detectLowerBlock() { - lowerBlockDetectorSubscription = lowerBoundBlockDetector.lowerBlock() + lowerBlockDetectorSubscription = lowerBoundService.detectLowerBounds() .subscribe { sendUpstreamStateEvent(UPDATED) } 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 76bc1466..cf1c9b80 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -31,13 +31,13 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Lifecycle -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.reactivestreams.Publisher @@ -150,8 +150,8 @@ class BitcoinGrpcUpstream( return true } - override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData { - return LowerBoundBlockDetector.LowerBlockData.default() + override fun getLowerBounds(): Collection { + return emptyList() } override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt index 3f535c4a..cee83d3c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt @@ -31,11 +31,11 @@ import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Lifecycle -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.domain.BlockHash import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import reactor.core.publisher.Flux import reactor.core.scheduler.Scheduler @@ -179,8 +179,8 @@ open class GenericGrpcUpstream( return true } - override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData { - return LowerBoundBlockDetector.LowerBlockData.default() + override fun getLowerBounds(): Collection { + return emptyList() } override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt new file mode 100644 index 00000000..d0c5aa65 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt @@ -0,0 +1,23 @@ +package io.emeraldpay.dshackle.upstream.lowerbound + +import java.time.Instant + +data class LowerBoundData( + val lowerBound: Long, + val timestamp: Long, + val type: LowerBoundType, +) : Comparable { + constructor(lowerBound: Long, type: LowerBoundType) : this(lowerBound, Instant.now().epochSecond, type) + + companion object { + fun default() = LowerBoundData(0, 0, LowerBoundType.UNKNOWN) + } + + override fun compareTo(other: LowerBoundData): Int { + return this.lowerBound.compareTo(other.lowerBound) + } +} + +enum class LowerBoundType { + UNKNOWN, STATE, SLOT +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundDetector.kt new file mode 100644 index 00000000..81c21742 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundDetector.kt @@ -0,0 +1,45 @@ +package io.emeraldpay.dshackle.upstream.lowerbound + +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import java.time.Duration +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean + +fun Long.toHex() = "0x${this.toString(16)}" + +abstract class LowerBoundDetector { + protected val log = LoggerFactory.getLogger(this::class.java) + + private val lowerBounds = ConcurrentHashMap() + + fun detectLowerBound(): Flux { + val notProcessing = AtomicBoolean(true) + + return Flux.interval( + Duration.ofSeconds(15), + Duration.ofMinutes(period()), + ) + .filter { notProcessing.get() } + .flatMap { + notProcessing.set(false) + internalDetectLowerBound() + .onErrorResume { Mono.just(LowerBoundData.default()) } + .switchIfEmpty(Flux.just(LowerBoundData.default())) + .doFinally { notProcessing.set(true) } + } + .filter { + it.lowerBound >= (lowerBounds[it.type]?.lowerBound ?: 0) + } + .map { + lowerBounds[it.type] = it + it + } + } + + // in minutes + protected abstract fun period(): Long + + protected abstract fun internalDetectLowerBound(): Flux +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundService.kt new file mode 100644 index 00000000..13ca03be --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundService.kt @@ -0,0 +1,32 @@ +package io.emeraldpay.dshackle.upstream.lowerbound + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import java.util.concurrent.ConcurrentHashMap + +typealias LowerBoundServiceBuilder = (Chain, Upstream) -> LowerBoundService + +abstract class LowerBoundService( + private val chain: Chain, + private val upstream: Upstream, +) { + private val log = LoggerFactory.getLogger(this::class.java) + + private val lowerBounds = ConcurrentHashMap() + + fun detectLowerBounds(): Flux { + return Flux.merge( + detectors().map { it.detectLowerBound() }, + ) + .doOnNext { + log.info("Lower bound of type ${it.type} is ${it.lowerBound} for upstream ${upstream.getId()} of chain $chain") + lowerBounds[it.type] = it + } + } + + fun getLowerBounds(): Collection = lowerBounds.values + + protected abstract fun detectors(): List +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt new file mode 100644 index 00000000..d19b844b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt @@ -0,0 +1,103 @@ +package io.emeraldpay.dshackle.upstream.lowerbound.detector + +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.kotlin.core.publisher.toFlux +import reactor.util.retry.Retry +import reactor.util.retry.RetryBackoffSpec +import java.time.Duration + +class RecursiveLowerBound( + private val upstream: Upstream, + private val type: LowerBoundType, + private val nonRetryableErrors: Set, +) { + private val log = LoggerFactory.getLogger(this::class.java) + + fun recursiveDetectLowerBound(hasData: (Long) -> Mono): Flux { + return Mono.just(upstream.getHead()) + .flatMap { + val currentHeight = it.getCurrentHeight() + if (currentHeight == null) { + Mono.empty() + } else { + Mono.just(LowerBoundBinarySearch(0, currentHeight)) + } + } + .expand { data -> + println(data) + if (data.found) { + Mono.empty() + } else { + val middle = middleBlock(data) + + if (data.left > data.right) { + val current = if (data.current == 0L) 1 else data.current + Mono.just(LowerBoundBinarySearch(current, true)) + } else { + hasData(middle) + .retryWhen(retrySpec(nonRetryableErrors)) + .flatMap(ChainResponse::requireResult) + .map { true } + .onErrorReturn(false) + .map { + if (it) { + LowerBoundBinarySearch(data.left, middle - 1, middle) + } else { + LowerBoundBinarySearch( + middle + 1, + data.right, + data.current, + ) + } + } + } + } + } + .filter { it.found } + .next() + .map { + LowerBoundData(it.current, type) + }.toFlux() + } + + private fun retrySpec(nonRetryableErrors: Set): RetryBackoffSpec { + return Retry.backoff( + Long.MAX_VALUE, + Duration.ofSeconds(1), + ) + .maxBackoff(Duration.ofMinutes(3)) + .filter { + !nonRetryableErrors.any { err -> it.message?.contains(err, true) ?: false } + } + .doAfterRetry { + log.debug( + "Error in calculation of lower block of upstream {}, retry attempt - {}, message - {}", + upstream.getId(), + it.totalRetries(), + it.failure().message, + ) + } + } + + private fun middleBlock(lowerBoundBinarySearch: LowerBoundBinarySearch): Long = + lowerBoundBinarySearch.left + (lowerBoundBinarySearch.right - lowerBoundBinarySearch.left) / 2 + + private data class LowerBoundBinarySearch( + val left: Long, + val right: Long, + val current: Long, + val found: Boolean, + ) { + constructor(left: Long, right: Long) : this(left, right, 0, false) + + constructor(left: Long, right: Long, current: Long) : this(left, right, current, false) + + constructor(current: Long, found: Boolean) : this(0, 0, current, found) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt index 394975bf..8256af6d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt @@ -9,7 +9,6 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.foundation.ChainOptions.Options import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.SingleCallValidator import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -17,6 +16,7 @@ import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import io.emeraldpay.dshackle.upstream.rpcclient.ObjectParams import java.math.BigInteger @@ -70,8 +70,8 @@ object NearChainSpecific : AbstractPollChainSpecific() { ) } - override fun lowerBoundBlockDetector(chain: Chain, upstream: Upstream): LowerBoundBlockDetector { - return NearLowerBoundBlockDetector(chain, upstream) + override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { + return NearLowerBoundService(chain, upstream) } fun validate(data: ByteArray): UpstreamAvailability { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundBlockDetector.kt deleted file mode 100644 index 1e06d3d5..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundBlockDetector.kt +++ /dev/null @@ -1,26 +0,0 @@ -package io.emeraldpay.dshackle.upstream.near - -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.rpcclient.ListParams -import reactor.core.publisher.Mono - -class NearLowerBoundBlockDetector( - chain: Chain, - val upstream: Upstream, -) : LowerBoundBlockDetector(chain, upstream) { - - override fun lowerBlockDetect(): Mono { - return upstream.getIngressReader().read(ChainRequest("status", ListParams())).map { - val resp = Global.objectMapper.readValue(it.getResult(), NearStatus::class.java) - LowerBlockData(resp.syncInfo.earliestHeight, null, resp.syncInfo.earliestBlockTime.toEpochMilli()) - } - } - - override fun periodRequest(): Long { - return 3 - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundService.kt new file mode 100644 index 00000000..76c5f3c4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundService.kt @@ -0,0 +1,15 @@ +package io.emeraldpay.dshackle.upstream.near + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService + +class NearLowerBoundService( + chain: Chain, + private val upstream: Upstream, +) : LowerBoundService(chain, upstream) { + override fun detectors(): List { + return listOf(NearLowerBoundStateDetector(upstream)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundStateDetector.kt new file mode 100644 index 00000000..3cd85770 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundStateDetector.kt @@ -0,0 +1,27 @@ +package io.emeraldpay.dshackle.upstream.near + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import reactor.core.publisher.Flux +import reactor.kotlin.core.publisher.toFlux + +class NearLowerBoundStateDetector( + private val upstream: Upstream, +) : LowerBoundDetector() { + + override fun period(): Long { + return 3 + } + + override fun internalDetectLowerBound(): Flux { + return upstream.getIngressReader().read(ChainRequest("status", ListParams())).map { + val resp = Global.objectMapper.readValue(it.getResult(), NearStatus::class.java) + LowerBoundData(resp.syncInfo.earliestHeight, LowerBoundType.STATE) + }.toFlux() + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt index 5d8f3a46..ffdc75bd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt @@ -15,7 +15,6 @@ import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription import io.emeraldpay.dshackle.upstream.LogsOracle -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.SingleCallValidator import io.emeraldpay.dshackle.upstream.Upstream @@ -29,6 +28,7 @@ import io.emeraldpay.dshackle.upstream.generic.GenericEgressSubscription import io.emeraldpay.dshackle.upstream.generic.GenericIngressSubscription import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator import io.emeraldpay.dshackle.upstream.generic.LocalReader +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.slf4j.LoggerFactory import reactor.core.publisher.Mono @@ -105,8 +105,8 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() { ) } - override fun lowerBoundBlockDetector(chain: Chain, upstream: Upstream): LowerBoundBlockDetector { - return PolkadotLowerBoundBlockDetector(chain, upstream) + override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { + return PolkadotLowerBoundService(chain, upstream) } fun validate(data: ByteArray, peers: Int, upstreamId: String): UpstreamAvailability { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt deleted file mode 100644 index 731116fc..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt +++ /dev/null @@ -1,47 +0,0 @@ -package io.emeraldpay.dshackle.upstream.polkadot - -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.ChainResponse -import io.emeraldpay.dshackle.upstream.RecursiveLowerBoundBlockDetector -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.rpcclient.ListParams -import io.emeraldpay.dshackle.upstream.toHex -import reactor.core.publisher.Mono - -class PolkadotLowerBoundBlockDetector( - chain: Chain, - private val upstream: Upstream, -) : RecursiveLowerBoundBlockDetector(chain, upstream) { - - companion object { - private val nonRetryableErrors = setOf( - "State already discarded for", - ) - } - - override fun hasState(blockNumber: Long): Mono { - return upstream.getIngressReader().read( - ChainRequest( - "chain_getBlockHash", - ListParams(blockNumber.toHex()), // in polkadot state methods work only with hash - ), - ) - .flatMap(ChainResponse::requireResult) - .map { - String(it, 1, it.size - 2) - } - .flatMap { - upstream.getIngressReader().read( - ChainRequest( - "state_getMetadata", - ListParams(it), - ), - ) - } - .retryWhen(retrySpec(nonRetryableErrors)) - .flatMap(ChainResponse::requireResult) - .map { true } - .onErrorReturn(false) - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundService.kt new file mode 100644 index 00000000..0ffc12f2 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundService.kt @@ -0,0 +1,16 @@ +package io.emeraldpay.dshackle.upstream.polkadot + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService + +class PolkadotLowerBoundService( + chain: Chain, + private val upstream: Upstream, +) : LowerBoundService(chain, upstream) { + + override fun detectors(): List { + return listOf(PolkadotLowerBoundStateDetector(upstream)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundStateDetector.kt new file mode 100644 index 00000000..384b7560 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundStateDetector.kt @@ -0,0 +1,51 @@ +package io.emeraldpay.dshackle.upstream.polkadot + +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound +import io.emeraldpay.dshackle.upstream.lowerbound.toHex +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import reactor.core.publisher.Flux + +class PolkadotLowerBoundStateDetector( + private val upstream: Upstream, +) : LowerBoundDetector() { + private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, nonRetryableErrors) + + companion object { + private val nonRetryableErrors = setOf( + "State already discarded for", + ) + } + + override fun period(): Long { + return 5 + } + + override fun internalDetectLowerBound(): Flux { + return recursiveLowerBound.recursiveDetectLowerBound { block -> + upstream.getIngressReader().read( + ChainRequest( + "chain_getBlockHash", + ListParams(block.toHex()), // in polkadot state methods work only with hash + ), + ) + .flatMap(ChainResponse::requireResult) + .map { + String(it, 1, it.size - 2) + } + .flatMap { + upstream.getIngressReader().read( + ChainRequest( + "state_getMetadata", + ListParams(it), + ), + ) + } + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt index 1c5282ce..f650a723 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt @@ -13,7 +13,6 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.DefaultSolanaMethods import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.IngressSubscription -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.SingleCallValidator import io.emeraldpay.dshackle.upstream.Upstream @@ -25,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.generic.AbstractChainSpecific import io.emeraldpay.dshackle.upstream.generic.GenericEgressSubscription import io.emeraldpay.dshackle.upstream.generic.GenericIngressSubscription import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.slf4j.LoggerFactory import reactor.core.publisher.Mono @@ -138,8 +138,8 @@ object SolanaChainSpecific : AbstractChainSpecific() { ) } - override fun lowerBoundBlockDetector(chain: Chain, upstream: Upstream): LowerBoundBlockDetector { - return SolanaLowerBoundBlockDetector(chain, upstream) + override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { + return SolanaLowerBoundService(chain, upstream) } override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundService.kt new file mode 100644 index 00000000..10763ba7 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundService.kt @@ -0,0 +1,15 @@ +package io.emeraldpay.dshackle.upstream.solana + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService + +class SolanaLowerBoundService( + chain: Chain, + private val upstream: Upstream, +) : LowerBoundService(chain, upstream) { + override fun detectors(): List { + return listOf(SolanaLowerBoundSlotDetector(upstream)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundSlotDetector.kt similarity index 77% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundSlotDetector.kt index b876d657..2623a9d7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundSlotDetector.kt @@ -1,24 +1,29 @@ package io.emeraldpay.dshackle.upstream.solana -import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.util.retry.Retry import java.time.Duration import kotlin.math.max -class SolanaLowerBoundBlockDetector( - chain: Chain, +class SolanaLowerBoundSlotDetector( private val upstream: Upstream, -) : LowerBoundBlockDetector(chain, upstream) { +) : LowerBoundDetector() { private val reader = upstream.getIngressReader() - override fun lowerBlockDetect(): Mono { + override fun period(): Long { + return 3 + } + + override fun internalDetectLowerBound(): Flux { return Mono.just(reader) .flatMap { it.read( @@ -34,7 +39,7 @@ class SolanaLowerBoundBlockDetector( slot } } - .flatMap { + .flatMapMany { reader.read( ChainRequest( "getBlock", // since getFirstAvailableBlock returns the slot of the lowest confirmed block we can directly call getBlock @@ -49,9 +54,12 @@ class SolanaLowerBoundBlockDetector( ), ) .flatMap(ChainResponse::requireResult) - .map { blockData -> + .flatMapMany { blockData -> val block = Global.objectMapper.readValue(blockData, SolanaBlock::class.java) - LowerBlockData(max(block.height, 1), it) + Flux.just( + LowerBoundData(max(block.height, 1), LowerBoundType.STATE), + LowerBoundData(it, LowerBoundType.SLOT), + ) } } .retryWhen( @@ -68,8 +76,4 @@ class SolanaLowerBoundBlockDetector( }, ) } - - override fun periodRequest(): Long { - return 3 - } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt index 0936ff1c..da582e4d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt @@ -9,13 +9,13 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.foundation.ChainOptions.Options import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.SingleCallValidator import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.slf4j.LoggerFactory import java.math.BigInteger @@ -71,8 +71,8 @@ object StarknetChainSpecific : AbstractPollChainSpecific() { ) } - override fun lowerBoundBlockDetector(chain: Chain, upstream: Upstream): LowerBoundBlockDetector { - return StarknetLowerBoundBlockDetector(chain, upstream) + override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { + return StarknetLowerBoundService(chain, upstream) } fun validate(data: ByteArray, lagging: Int, upstreamId: String): UpstreamAvailability { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetector.kt deleted file mode 100644 index a86fab28..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetector.kt +++ /dev/null @@ -1,21 +0,0 @@ -package io.emeraldpay.dshackle.upstream.starknet - -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector -import io.emeraldpay.dshackle.upstream.Upstream -import reactor.core.publisher.Mono - -class StarknetLowerBoundBlockDetector( - chain: Chain, - upstream: Upstream, -) : LowerBoundBlockDetector(chain, upstream) { - - // for starknet we assume that all nodes are archive - override fun lowerBlockDetect(): Mono { - return Mono.just(LowerBlockData(1)) - } - - override fun periodRequest(): Long { - return 120 - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundService.kt new file mode 100644 index 00000000..e228eb22 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundService.kt @@ -0,0 +1,15 @@ +package io.emeraldpay.dshackle.upstream.starknet + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService + +class StarknetLowerBoundService( + chain: Chain, + upstream: Upstream, +) : LowerBoundService(chain, upstream) { + override fun detectors(): List { + return listOf(StarknetLowerBoundStateDetector()) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetector.kt new file mode 100644 index 00000000..983870db --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetector.kt @@ -0,0 +1,17 @@ +package io.emeraldpay.dshackle.upstream.starknet + +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import reactor.core.publisher.Flux + +class StarknetLowerBoundStateDetector : LowerBoundDetector() { + + override fun period(): Long { + return 120 + } + + override fun internalDetectLowerBound(): Flux { + return Flux.just(LowerBoundData(1, LowerBoundType.STATE)) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy index 88124754..2ae39c26 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy @@ -23,12 +23,13 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.* import io.emeraldpay.dshackle.upstream.generic.GenericUpstream -import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import org.jetbrains.annotations.NotNull import org.reactivestreams.Publisher @@ -75,7 +76,7 @@ class GenericUpstreamMock extends GenericUpstream { new ConnectorFactoryMock(api, new EthereumHeadMock()), io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&validator, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&upstreamSettingsDetector, - io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&lowerBoundBlockDetector, + io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&lowerBoundService, ) this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) @@ -108,7 +109,7 @@ class GenericUpstreamMock extends GenericUpstream { } @Override - LowerBoundBlockDetector.LowerBlockData getLowerBlock() { - return new LowerBoundBlockDetector.LowerBlockData(0, 0) + Collection getLowerBounds() { + return List.of(new LowerBoundData(0, LowerBoundType.STATE)) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 4c63344f..12564741 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -77,7 +77,7 @@ class FilteredApisSpec extends Specification { connectorFactory, cs.&validator, cs.&upstreamSettingsDetector, - cs.&lowerBoundBlockDetector + cs.&lowerBoundService ) } def matcher = new Selector.LabelMatcher("test", ["foo"]) diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt similarity index 77% rename from src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt rename to src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt index 43d79b51..7aa18f12 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt @@ -2,11 +2,15 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.reader.ChainReader -import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundBlockDetector +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundService import io.emeraldpay.dshackle.upstream.ethereum.ZERO_ADDRESS -import io.emeraldpay.dshackle.upstream.polkadot.PolkadotLowerBoundBlockDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.lowerbound.toHex +import io.emeraldpay.dshackle.upstream.polkadot.PolkadotLowerBoundService import io.emeraldpay.dshackle.upstream.rpcclient.ListParams -import org.junit.jupiter.api.Assertions.assertEquals +import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.Arguments import org.junit.jupiter.params.provider.MethodSource @@ -17,13 +21,13 @@ import reactor.core.publisher.Mono import reactor.test.StepVerifier import java.time.Duration -class RecursiveLowerBoundBlockDetectorTest { +class RecursiveLowerBoundServiceTest { @ParameterizedTest @MethodSource("detectors") fun `find lower block closer to the height`( reader: ChainReader, - detectorClass: Class, + detectorClass: Class, ) { val head = mock { on { getCurrentHeight() } doReturn 18000000 @@ -35,21 +39,25 @@ class RecursiveLowerBoundBlockDetectorTest { val detector = detectorClass.getConstructor(Chain::class.java, Upstream::class.java).newInstance(Chain.UNSPECIFIED, upstream) - StepVerifier.withVirtualTime { detector.lowerBlock() } + StepVerifier.withVirtualTime { detector.detectLowerBounds() } .expectSubscription() .expectNoEvent(Duration.ofSeconds(15)) - .expectNextMatches { it.blockNumber == 17964844L } + .expectNextMatches { it.lowerBound == 17964844L && it.type == LowerBoundType.STATE } .thenCancel() .verify(Duration.ofSeconds(3)) - assertEquals(17964844L, detector.getCurrentLowerBlock().blockNumber) + assertThat(detector.getLowerBounds().toList()) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields("timestamp") + .hasSameElementsAs( + listOf(LowerBoundData(17964844L, LowerBoundType.STATE)), + ) } @ParameterizedTest @MethodSource("detectorsFirstBlock") fun `lower block is 0x1`( reader: ChainReader, - detectorClass: Class, + detectorClass: Class, ) { val head = mock { on { getCurrentHeight() } doReturn 18000000 @@ -61,14 +69,18 @@ class RecursiveLowerBoundBlockDetectorTest { val detector = detectorClass.getConstructor(Chain::class.java, Upstream::class.java).newInstance(Chain.UNSPECIFIED, upstream) - StepVerifier.withVirtualTime { detector.lowerBlock() } + StepVerifier.withVirtualTime { detector.detectLowerBounds() } .expectSubscription() .expectNoEvent(Duration.ofSeconds(15)) - .expectNextMatches { it.blockNumber == 1L } + .expectNextMatches { it.lowerBound == 1L } .thenCancel() .verify(Duration.ofSeconds(3)) - assertEquals(1, detector.getCurrentLowerBlock().blockNumber) + assertThat(detector.getLowerBounds().toList()) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields("timestamp") + .hasSameElementsAs( + listOf(LowerBoundData(1L, LowerBoundType.STATE)), + ) } companion object { @@ -96,7 +108,7 @@ class RecursiveLowerBoundBlockDetectorTest { } } }, - EthereumLowerBoundBlockDetector::class.java, + EthereumLowerBoundService::class.java, ), Arguments.of( mock { @@ -118,7 +130,7 @@ class RecursiveLowerBoundBlockDetectorTest { } } }, - PolkadotLowerBoundBlockDetector::class.java, + PolkadotLowerBoundService::class.java, ), ) @@ -130,13 +142,13 @@ class RecursiveLowerBoundBlockDetectorTest { read(any()) } doReturn Mono.just(ChainResponse("\"0x1\"".toByteArray(), null)) }, - PolkadotLowerBoundBlockDetector::class.java, + PolkadotLowerBoundService::class.java, ), Arguments.of( mock { on { read(any()) } doReturn Mono.just(ChainResponse(ByteArray(0), null)) }, - EthereumLowerBoundBlockDetector::class.java, + EthereumLowerBoundService::class.java, ), ) } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundServiceTest.kt similarity index 69% rename from src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt rename to src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundServiceTest.kt index 851de30e..6df90e15 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundServiceTest.kt @@ -6,8 +6,10 @@ import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import io.emeraldpay.dshackle.upstream.rpcclient.ListParams -import org.junit.jupiter.api.Assertions.assertEquals +import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.api.Test import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock @@ -15,7 +17,7 @@ import reactor.core.publisher.Mono import reactor.test.StepVerifier import java.time.Duration -class SolanaLowerBoundBlockDetectorTest { +class SolanaLowerBoundServiceTest { @Test fun `get solana lower block and slot`() { @@ -54,16 +56,23 @@ class SolanaLowerBoundBlockDetectorTest { on { getIngressReader() } doReturn reader } - val detector = SolanaLowerBoundBlockDetector(Chain.UNSPECIFIED, upstream) + val detector = SolanaLowerBoundService(Chain.UNSPECIFIED, upstream) - StepVerifier.withVirtualTime { detector.lowerBlock() } + StepVerifier.withVirtualTime { detector.detectLowerBounds() } .expectSubscription() .expectNoEvent(Duration.ofSeconds(15)) - .expectNextMatches { it.blockNumber == 21000000L && it.slot == 25000000L } + .expectNextMatches { it.lowerBound == 21000000L && it.type == LowerBoundType.STATE } + .expectNextMatches { it.lowerBound == 25000000L && it.type == LowerBoundType.SLOT } .thenCancel() .verify(Duration.ofSeconds(3)) - assertEquals(21000000, detector.getCurrentLowerBlock().blockNumber) - assertEquals(25000000, detector.getCurrentLowerBlock().slot) + assertThat(detector.getLowerBounds().toList()) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields("timestamp") + .hasSameElementsAs( + listOf( + LowerBoundData(21000000L, LowerBoundType.STATE), + LowerBoundData(25000000L, LowerBoundType.SLOT), + ), + ) } } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetectorTest.kt deleted file mode 100644 index 456b1452..00000000 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetectorTest.kt +++ /dev/null @@ -1,24 +0,0 @@ -package io.emeraldpay.dshackle.upstream.starknet - -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector -import io.emeraldpay.dshackle.upstream.Upstream -import org.junit.jupiter.api.Test -import org.mockito.kotlin.mock -import reactor.test.StepVerifier -import java.time.Duration - -class StarknetLowerBoundBlockDetectorTest { - - @Test - fun `starknet lower block is 1`() { - val detector = StarknetLowerBoundBlockDetector(Chain.UNSPECIFIED, mock()) - - StepVerifier.withVirtualTime { detector.lowerBlock() } - .expectSubscription() - .expectNoEvent(Duration.ofSeconds(15)) - .expectNext(LowerBoundBlockDetector.LowerBlockData(1)) - .thenCancel() - .verify(Duration.ofSeconds(3)) - } -} diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetectorTest.kt new file mode 100644 index 00000000..81ba1845 --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetectorTest.kt @@ -0,0 +1,22 @@ +package io.emeraldpay.dshackle.upstream.starknet + +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import org.junit.jupiter.api.Test +import reactor.test.StepVerifier +import java.time.Duration + +class StarknetLowerBoundStateDetectorTest { + + @Test + fun `starknet lower block is 1`() { + val detector = StarknetLowerBoundStateDetector() + + StepVerifier.withVirtualTime { detector.detectLowerBound() } + .expectSubscription() + .expectNoEvent(Duration.ofSeconds(15)) + .expectNext(LowerBoundData(1, LowerBoundType.STATE)) + .thenCancel() + .verify(Duration.ofSeconds(3)) + } +}