StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactoring lower bounds (#450)

Browse Source
This commit is contained in:
KirillPamPam
2024-04-11 18:03:32 +04:00
committed by GitHub
parent 0113862ac4
commit 55e2dc70c4
44 changed files with 619 additions and 418 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
View File

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

45
src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
View File

@@ -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<LowerBoundData>): 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,
)
}

2
src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt
View File

@@ -86,7 +86,7 @@ open class GenericUpstreamCreator(
connectorFactory,
cs::validator,
cs::upstreamSettingsDetector,
cs::lowerBoundBlockDetector,
cs::lowerBoundService,
)
upstream.start()

87
src/main/kotlin/io/emeraldpay/dshackle/upstream/LowerBoundBlockDetector.kt
View File

@@ -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<LowerBlockData> {
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<LowerBlockData>
protected abstract fun periodRequest(): Long
data class LowerBlockData(
val blockNumber: Long,
val slot: Long?,
val timestamp: Long,
) : Comparable<LowerBlockData> {
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)
}
}

24
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -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<Capability> = emptySet()
@Volatile
private var lowerBlock: LowerBoundBlockDetector.LowerBlockData = LowerBoundBlockDetector.LowerBlockData.default()
private val lowerBounds = ConcurrentHashMap<LowerBoundType, LowerBoundData>()
@Volatile
private var quorumLabels: List<QuorumForLabels.QuorumItem>? = 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<LowerBoundData> {
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 {

79
src/main/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetector.kt
View File

@@ -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<LowerBlockData> {
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<String>): 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<Boolean>
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
View File

@@ -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<Capability>
fun isGrpc(): Boolean
fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData
fun getLowerBounds(): Collection<LowerBoundData>
fun getUpstreamSettingsData(): UpstreamSettingsData?
fun <T : Upstream> cast(selfType: Class<T>): T

21
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundBlockDetector.kt
View File

@@ -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<LowerBlockData> {
return Mono.just(LowerBlockData(1))
}
override fun periodRequest(): Long {
return 120
}
}

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt Normal file
View File

@@ -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<LowerBoundDetector> {
return listOf(BeaconChainLowerBoundStateDetector())
}
}

17
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt Normal file
View File

@@ -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<LowerBoundData> {
return Flux.just(LowerBoundData(1, LowerBoundType.STATE))
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt
View File

@@ -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)
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
View File

@@ -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<LowerBoundData> {
return emptyList()
}
override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt
View File

@@ -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 {

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt Normal file
View File

@@ -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<LowerBoundDetector> {
return listOf(EthereumLowerBoundStateDetector(upstream))
}
}

44
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt
View File

@@ -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<Boolean> {
if (blockNumber == 0L) {
return Mono.just(true)
override fun period(): Long {
return 5
}
override fun internalDetectLowerBound(): Flux<LowerBoundData> {
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)
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
View File

@@ -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 {

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
View File

@@ -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<LowerBoundData> {
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)
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -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<LowerBoundData> {
return emptyList()
}
override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt
View File

@@ -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<LowerBoundData> {
return emptyList()
}
override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt Normal file
View File

@@ -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<LowerBoundData> {
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
}

45
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundDetector.kt Normal file
View File

@@ -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<LowerBoundType, LowerBoundData>()
fun detectLowerBound(): Flux<LowerBoundData> {
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<LowerBoundData>
}

32
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundService.kt Normal file
View File

@@ -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<LowerBoundType, LowerBoundData>()
fun detectLowerBounds(): Flux<LowerBoundData> {
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<LowerBoundData> = lowerBounds.values
protected abstract fun detectors(): List<LowerBoundDetector>
}

103
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt Normal file
View File

@@ -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<String>,
) {
private val log = LoggerFactory.getLogger(this::class.java)
fun recursiveDetectLowerBound(hasData: (Long) -> Mono<ChainResponse>): Flux<LowerBoundData> {
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<String>): 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)
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt
View File

@@ -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 {

26
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundBlockDetector.kt
View File

@@ -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<LowerBlockData> {
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
}
}

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundService.kt Normal file
View File

@@ -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<LowerBoundDetector> {
return listOf(NearLowerBoundStateDetector(upstream))
}
}

27
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundStateDetector.kt Normal file
View File

@@ -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<LowerBoundData> {
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()
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt
View File

@@ -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 {

47
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt
View File

@@ -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<Boolean> {
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)
}
}

16
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundService.kt Normal file
View File

@@ -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<LowerBoundDetector> {
return listOf(PolkadotLowerBoundStateDetector(upstream))
}
}

51
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundStateDetector.kt Normal file
View File

@@ -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<LowerBoundData> {
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),
),
)
}
}
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt
View File

@@ -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 {

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundService.kt Normal file
View File

@@ -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<LowerBoundDetector> {
return listOf(SolanaLowerBoundSlotDetector(upstream))
}
}

30
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundSlotDetector.kt
View File

@@ -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<LowerBlockData> {
override fun period(): Long {
return 3
}
override fun internalDetectLowerBound(): Flux<LowerBoundData> {
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
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt
View File

@@ -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 {

21
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetector.kt
View File

@@ -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<LowerBlockData> {
return Mono.just(LowerBlockData(1))
}
override fun periodRequest(): Long {
return 120
}
}

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundService.kt Normal file
View File

@@ -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<LowerBoundDetector> {
return listOf(StarknetLowerBoundStateDetector())
}
}

17
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetector.kt Normal file
View File

@@ -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<LowerBoundData> {
return Flux.just(LowerBoundData(1, LowerBoundType.STATE))
}
}

13
src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy
View File

@@ -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<LowerBoundData> getLowerBounds() {
return List.of(new LowerBoundData(0, LowerBoundType.STATE))
}
}

2
src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
View File

@@ -77,7 +77,7 @@ class FilteredApisSpec extends Specification {
connectorFactory,
cs.&validator,
cs.&upstreamSettingsDetector,
cs.&lowerBoundBlockDetector
cs.&lowerBoundService
)
}
def matcher = new Selector.LabelMatcher("test", ["foo"])

44
src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundBlockDetectorTest.kt → src/test/kotlin/io/emeraldpay/dshackle/upstream/RecursiveLowerBoundServiceTest.kt
View File

@@ -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<LowerBoundBlockDetector>,
detectorClass: Class<LowerBoundService>,
) {
val head = mock<Head> {
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<LowerBoundBlockDetector>,
detectorClass: Class<LowerBoundService>,
) {
val head = mock<Head> {
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<ChainReader> {
@@ -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<ChainReader> {
on { read(any()) } doReturn Mono.just(ChainResponse(ByteArray(0), null))
},
EthereumLowerBoundBlockDetector::class.java,
EthereumLowerBoundService::class.java,
),
)
}

23
src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetectorTest.kt → src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundServiceTest.kt
View File

@@ -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),
),
)
}
}

24
src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundBlockDetectorTest.kt
View File

@@ -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<Upstream>())
StepVerifier.withVirtualTime { detector.lowerBlock() }
.expectSubscription()
.expectNoEvent(Duration.ofSeconds(15))
.expectNext(LowerBoundBlockDetector.LowerBlockData(1))
.thenCancel()
.verify(Duration.ofSeconds(3))
}
}

22
src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetectorTest.kt Normal file
View File

@@ -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))
}
}