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

Predict lower height and use lower height matcher (#552)

Browse Source
This commit is contained in:
KirillPamPam
2024-08-22 19:52:57 +04:00
committed by GitHub
parent 32cb726aab
commit e184bb3547
30 changed files with 582 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -168,6 +168,10 @@ abstract class DefaultUpstream(
// NOOP
}
override fun predictLowerBound(type: LowerBoundType): Long {
return 0
}
protected fun sendUpstreamStateEvent(eventType: UpstreamChangeEvent.ChangeType) {
stateEventStream.emitNext(
UpstreamChangeEvent(chain, this, eventType),

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/MatchesResponse.kt
View File

@@ -1,5 +1,7 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
sealed class MatchesResponse {
fun matched(): Boolean {
@@ -25,6 +27,13 @@ sealed class MatchesResponse {
.joinToString("; ") { it.getCause()!! }
is NotMatchedResponse -> "Not matched - ${response.getCause()}"
is SameNodeResponse -> "Upstream does not have hash ${this.upstreamHash}"
is LowerHeightResponse -> {
if (this.predictedHeight == 0L) {
"Upstream lower height of type ${this.boundType} cannot be predicted"
} else {
"Upstream lower height ${this.predictedHeight} of type ${this.boundType} is greater than ${this.lowerHeight}"
}
}
else -> null
}
@@ -71,6 +80,12 @@ sealed class MatchesResponse {
object GrpcResponse : MatchesResponse()
data class LowerHeightResponse(
val lowerHeight: Long,
val predictedHeight: Long,
val boundType: LowerBoundType,
) : MatchesResponse()
data class HeightResponse(
val height: Long,
val currentHeight: Long,

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

@@ -257,6 +257,10 @@ abstract class Multistream(
return getAll().any { it.isAvailable() }
}
override fun predictLowerBound(type: LowerBoundType): Long {
return 0
}
override fun getStatus(): UpstreamAvailability {
return state.getStatus()
}

41
src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
View File

@@ -23,11 +23,13 @@ import io.emeraldpay.dshackle.upstream.MatchesResponse.CapabilityResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.ExistsResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.GrpcResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.HeightResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.LowerHeightResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.NotMatchedResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.SameNodeResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.SlotHeightResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.Success
import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
import io.emeraldpay.dshackle.upstream.lowerbound.fromProtoType
import org.apache.commons.lang3.StringUtils
import java.util.Collections
@@ -101,6 +103,16 @@ class Selector {
else -> empty
}
}
it.hasLowerHeightSelector() -> {
if (it.lowerHeightSelector.height > 0) {
LowerHeightMatcher(
it.lowerHeightSelector.height,
it.lowerHeightSelector.lowerBoundType.fromProtoType(),
)
} else {
empty
}
}
else -> empty
}
}.run {
@@ -112,8 +124,11 @@ class Selector {
private fun getSort(selectors: List<BlockchainOuterClass.Selector>): Sort {
selectors.forEach { selector ->
if (selector.hasHeightSelector()) {
return HeightNumberOrTag.fromHeightSelector(selector.heightSelector)?.getSort() ?: Sort.default
} else if (selector.hasLowerHeightSelector()) {
val heightSort = HeightNumberOrTag.fromHeightSelector(selector.heightSelector)?.getSort() ?: Sort.default
if (heightSort != Sort.default) {
return heightSort
}
} else if (selector.hasLowerHeightSelector() && selector.lowerHeightSelector.height == 0L) {
return Sort(
compareBy(nullsLast()) {
it.getLowerBound(selector.lowerHeightSelector.lowerBoundType.fromProtoType())?.lowerBound
@@ -546,6 +561,28 @@ class Selector {
}
}
data class LowerHeightMatcher(
private val lowerHeight: Long,
private val boundType: LowerBoundType,
) : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse {
val predictedLowerBound = up.predictLowerBound(boundType)
return if (lowerHeight >= predictedLowerBound && predictedLowerBound != 0L) {
Success
} else {
LowerHeightResponse(lowerHeight, predictedLowerBound, boundType)
}
}
override fun describeInternal(): String {
return "lower height $lowerHeight"
}
override fun toString(): String {
return "Matcher: ${describeInternal()}"
}
}
class HeightMatcher(val height: Long) : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse {

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

@@ -55,6 +55,7 @@ interface Upstream : Lifecycle {
fun addFinalization(finalization: FinalizationData, upstreamId: String)
fun getUpstreamSettingsData(): UpstreamSettingsData?
fun updateLowerBound(lowerBound: Long, type: LowerBoundType)
fun predictLowerBound(type: LowerBoundType): Long
fun getChain(): Chain

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundService.kt
View File

@@ -6,10 +6,10 @@ import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
class BeaconChainLowerBoundService(
chain: Chain,
private val chain: Chain,
upstream: Upstream,
) : LowerBoundService(chain, upstream) {
override fun detectors(): List<LowerBoundDetector> {
return listOf(BeaconChainLowerBoundStateDetector())
return listOf(BeaconChainLowerBoundStateDetector(chain))
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLowerBoundStateDetector.kt
View File

@@ -1,11 +1,14 @@
package io.emeraldpay.dshackle.upstream.beaconchain
import io.emeraldpay.dshackle.Chain
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() {
class BeaconChainLowerBoundStateDetector(
private val chain: Chain,
) : LowerBoundDetector(chain) {
override fun period(): Long {
return 120

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/cosmos/CosmosLowerBoundService.kt
View File

@@ -24,7 +24,7 @@ class CosmosLowerBoundService(
class CosmosLowerBoundStateDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
override fun period(): Long {
return 3

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt
View File

@@ -16,7 +16,7 @@ import reactor.core.publisher.Mono
class EthereumLowerBoundBlockDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
companion object {
private const val NO_BLOCK_DATA = "No block data"

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundLogsDetector.kt
View File

@@ -12,7 +12,7 @@ import reactor.core.publisher.Flux
class EthereumLowerBoundLogsDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
companion object {
const val MAX_OFFSET = 20

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundStateDetector.kt
View File

@@ -15,7 +15,7 @@ import reactor.core.publisher.Mono
class EthereumLowerBoundStateDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, stateErrors, lowerBounds)
companion object {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundTxDetector.kt
View File

@@ -13,7 +13,7 @@ import reactor.core.publisher.Flux
class EthereumLowerBoundTxDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
companion object {
const val MAX_OFFSET = 20

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

@@ -354,5 +354,9 @@ open class GenericUpstream(
lowerBoundService.updateLowerBound(lowerBound, type)
}
override fun predictLowerBound(type: LowerBoundType): Long {
return lowerBoundService.predictLowerBound(type)
}
fun isValid(): Boolean = isUpstreamValid.get()
}

16
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundDetector.kt
View File

@@ -1,19 +1,21 @@
package io.emeraldpay.dshackle.upstream.lowerbound
import io.emeraldpay.dshackle.Chain
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.Sinks
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 {
abstract class LowerBoundDetector(
chain: Chain,
) {
protected val log = LoggerFactory.getLogger(this::class.java)
protected val lowerBounds = ConcurrentHashMap<LowerBoundType, LowerBoundData>()
protected val lowerBounds = LowerBounds(chain)
private val lowerBoundSink = Sinks.many().multicast().directBestEffort<LowerBoundData>()
fun detectLowerBound(): Flux<LowerBoundData> {
@@ -35,10 +37,10 @@ abstract class LowerBoundDetector {
},
)
.filter {
it.lowerBound >= (lowerBounds[it.type]?.lowerBound ?: 0)
it.lowerBound >= (lowerBounds.getLastBound(it.type)?.lowerBound ?: 0)
}
.map {
lowerBounds[it.type] = it
lowerBounds.updateBound(it)
it
}
}
@@ -53,4 +55,8 @@ abstract class LowerBoundDetector {
fun updateLowerBound(lowerBound: Long, type: LowerBoundType) {
lowerBoundSink.emitNext(LowerBoundData(lowerBound, type)) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
}
fun predictLowerBound(type: LowerBoundType): Long {
return lowerBounds.predictNextBound(type)
}
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundService.kt
View File

@@ -33,6 +33,13 @@ abstract class LowerBoundService(
.forEach { it.updateLowerBound(lowerBound, type) }
}
fun predictLowerBound(type: LowerBoundType): Long {
return detectors
.firstOrNull { it.types().contains(type) }
?.predictLowerBound(type)
?: 0
}
fun getLowerBounds(): Collection<LowerBoundData> = lowerBounds.values
fun getLowerBound(lowerBoundType: LowerBoundType): LowerBoundData? = lowerBounds[lowerBoundType]

124
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBounds.kt Normal file
View File

@@ -0,0 +1,124 @@
package io.emeraldpay.dshackle.upstream.lowerbound
import com.google.common.util.concurrent.AtomicDouble
import io.emeraldpay.dshackle.Chain
import org.apache.commons.math3.stat.regression.SimpleRegression
import java.time.Instant
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedDeque
import kotlin.math.roundToLong
class LowerBounds(
chain: Chain,
) {
companion object {
private const val MAX_BOUNDS = 3
}
private val averageSpeed = chain.averageRemoveDataSpeed
private val lowerBounds = ConcurrentHashMap<LowerBoundType, LowerBoundCoeffs>()
fun updateBound(newBound: LowerBoundData) {
if (lowerBounds.containsKey(newBound.type)) {
val lowerBoundCoeffs = lowerBounds[newBound.type]!!
// we add only bounds with different timestamps
if (newBound.timestamp != lowerBoundCoeffs.getLastBound().timestamp) {
if (newBound.lowerBound == 1L) {
// this is the fully archival node, so there is no need to accumulate bounds and calculate the coeffs
lowerBoundCoeffs.updateCoeffs(0.0, 1.0)
lowerBoundCoeffs.clearBounds()
lowerBoundCoeffs.addBound(newBound)
} else {
// accumulate up to MAX_BOUNDS and preserve this size
if (lowerBoundCoeffs.boundsSize() == MAX_BOUNDS) {
lowerBoundCoeffs.removeFirst()
}
lowerBoundCoeffs.addBound(newBound)
if (lowerBoundCoeffs.boundsSize() < MAX_BOUNDS) {
// calculate coeffs based on the average speed until we accumulate al least MAX_BOUNDS bounds
lowerBoundCoeffs.updateCoeffs(averageSpeed, calculateB(newBound))
} else {
// having MAX_BOUNDS bounds we can use linear regression
lowerBoundCoeffs.train()
}
}
}
} else {
// add new bound if it hasn't existed yet
lowerBounds[newBound.type] = LowerBoundCoeffs()
.apply {
addBound(newBound)
if (newBound.lowerBound == 1L) {
// this is the fully archival node
updateCoeffs(0.0, 1.0)
} else {
// otherwise we calculate the coeffs based on the average speed
updateCoeffs(averageSpeed, calculateB(newBound))
}
}
}
}
fun predictNextBound(type: LowerBoundType): Long {
val lowerBoundCoeffs = lowerBounds[type] ?: return 0
val xTime = Instant.now().epochSecond
return (lowerBoundCoeffs.k.get() * xTime + lowerBoundCoeffs.b.get()).roundToLong()
}
fun getLastBound(type: LowerBoundType): LowerBoundData? {
return lowerBounds[type]?.getLastBound()
}
fun getAllBounds(type: LowerBoundType): List<LowerBoundData> {
return lowerBounds[type]?.lowerBounds?.toList() ?: emptyList()
}
private fun calculateB(bound: LowerBoundData): Double {
return bound.lowerBound.toDouble() - (averageSpeed * bound.timestamp)
}
// to predict the next lower bound we use linear regression, y = kx + b,
// where x - current time, y - the predicted bound, k and b - coefficients
// to achieve that we accumulate up to max bounds (3 by default) and then calculate the coefficients using the regression lib
// having these coeffs we can predict the next bound in the predictNextBound() method
private class LowerBoundCoeffs {
val lowerBounds = ConcurrentLinkedDeque<LowerBoundData>()
val k = AtomicDouble()
val b = AtomicDouble()
fun addBound(bound: LowerBoundData) {
lowerBounds.add(bound)
}
fun updateCoeffs(newK: Double, newB: Double) {
k.set(newK)
b.set(newB)
}
fun clearBounds() {
lowerBounds.clear()
}
fun removeFirst() {
lowerBounds.removeFirst()
}
fun boundsSize(): Int = lowerBounds.size
fun getLastBound(): LowerBoundData = lowerBounds.last
fun train() {
val regression = SimpleRegression()
lowerBounds.forEach {
regression.addObservation(doubleArrayOf(it.timestamp.toDouble()), it.lowerBound.toDouble())
}
updateCoeffs(regression.slope, regression.intercept)
}
}
}

17
src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt
View File

@@ -4,6 +4,7 @@ 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.lowerbound.LowerBounds
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@@ -17,7 +18,7 @@ class RecursiveLowerBound(
private val upstream: Upstream,
private val type: LowerBoundType,
private val nonRetryableErrors: Set<String>,
private val lowerBounds: Map<LowerBoundType, LowerBoundData>,
private val lowerBounds: LowerBounds,
) {
private val log = LoggerFactory.getLogger(this::class.java)
@@ -56,13 +57,13 @@ class RecursiveLowerBound(
fun recursiveDetectLowerBoundWithOffset(maxLimit: Int, hasData: (Long) -> Mono<ChainResponse>): Flux<LowerBoundData> {
val visitedBlocks = HashSet<Long>()
return Mono.justOrEmpty(lowerBounds[type]?.lowerBound)
.flatMapMany {
return Mono.justOrEmpty(lowerBounds.getLastBound(type)?.lowerBound)
.flatMapMany { bound ->
// at first, we try to check the current bound to prevent huge calculations
hasData(it!!)
.retryWhen(retrySpec(it, nonRetryableErrors))
hasData(bound!!)
.retryWhen(retrySpec(bound, nonRetryableErrors))
.flatMap(ChainResponse::requireResult)
.map { LowerBoundData(lowerBounds[type]!!.lowerBound, type) }
.map { LowerBoundData(bound, type) }
.onErrorResume { Mono.empty() }
}.switchIfEmpty(
initialRange()
@@ -150,11 +151,11 @@ class RecursiveLowerBound(
val currentHeight = it.getCurrentHeight()
if (currentHeight == null) {
Mono.empty()
} else if (!lowerBounds.contains(type)) {
} else if (lowerBounds.getLastBound(type) == null) {
Mono.just(LowerBoundBinarySearchData(0, currentHeight))
} else {
// next calculations will be carried out only within the last range
Mono.just(LowerBoundBinarySearchData(lowerBounds[type]!!.lowerBound, currentHeight))
Mono.just(LowerBoundBinarySearchData(lowerBounds.getLastBound(type)!!.lowerBound, currentHeight))
}
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundStateDetector.kt
View File

@@ -11,7 +11,7 @@ import reactor.core.publisher.Flux
class NearLowerBoundStateDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
override fun period(): Long {
return 3

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundStateDetector.kt
View File

@@ -13,7 +13,7 @@ import reactor.core.publisher.Flux
class PolkadotLowerBoundStateDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.STATE, nonRetryableErrors, lowerBounds)
companion object {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundSlotDetector.kt
View File

@@ -16,7 +16,7 @@ import kotlin.math.max
class SolanaLowerBoundSlotDetector(
private val upstream: Upstream,
) : LowerBoundDetector() {
) : LowerBoundDetector(upstream.getChain()) {
private val reader = upstream.getIngressReader()
override fun period(): Long {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundService.kt
View File

@@ -6,10 +6,10 @@ import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
class StarknetLowerBoundService(
chain: Chain,
private val chain: Chain,
upstream: Upstream,
) : LowerBoundService(chain, upstream) {
override fun detectors(): List<LowerBoundDetector> {
return listOf(StarknetLowerBoundStateDetector())
return listOf(StarknetLowerBoundStateDetector(chain))
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetLowerBoundStateDetector.kt
View File

@@ -1,11 +1,14 @@
package io.emeraldpay.dshackle.upstream.starknet
import io.emeraldpay.dshackle.Chain
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() {
class StarknetLowerBoundStateDetector(
chain: Chain,
) : LowerBoundDetector(chain) {
override fun period(): Long {
return 120