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

validators for starknet, vara and solana (#350)

Browse Source
This commit is contained in:
a10zn8
2023-11-26 16:42:43 +03:00
committed by GitHub
parent 312986c2b5
commit d0322a0dab
11 changed files with 280 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt
View File

@@ -3,15 +3,38 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
import io.emeraldpay.dshackle.foundation.ChainOptions
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration
typealias UpstreamValidatorBuilder = (Chain, Upstream, ChainOptions.Options, ChainConfig) -> UpstreamValidator?
interface UpstreamValidator {
fun start(): Flux<UpstreamAvailability>
abstract class UpstreamValidator(
val upstream: Upstream,
val options: ChainOptions.Options,
) {
companion object {
private val log = LoggerFactory.getLogger(UpstreamValidator::class.java)
}
fun start(): Flux<UpstreamAvailability> {
return Flux.interval(
Duration.ZERO,
Duration.ofSeconds(options.validationInterval.toLong()),
).subscribeOn(EthereumUpstreamValidator.scheduler)
.flatMap {
validate()
}
.doOnNext {
log.debug("Status after validation is {} for {}", it, upstream.getId())
}
}
fun validateUpstreamSettings(): Mono<ValidateUpstreamSettingsResult>
abstract fun validate(): Mono<UpstreamAvailability>
abstract fun validateUpstreamSettings(): Mono<ValidateUpstreamSettingsResult>
fun validateUpstreamSettingsOnStartup(): ValidateUpstreamSettingsResult {
return validateUpstreamSettings().block() ?: ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR
@@ -23,3 +46,8 @@ enum class ValidateUpstreamSettingsResult {
UPSTREAM_SETTINGS_ERROR,
UPSTREAM_FATAL_SETTINGS_ERROR,
}
data class SingleCallValidator(
val method: JsonRpcRequest,
val check: (ByteArray) -> UpstreamAvailability,
)

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

@@ -84,11 +84,11 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
upstream: Upstream,
options: Options,
config: ChainConfig,
): UpstreamValidator? {
): UpstreamValidator {
return EthereumUpstreamValidator(chain, upstream, options, config)
}
override fun labelDetector(chain: Chain, reader: JsonRpcReader): LabelsDetector? {
override fun labelDetector(chain: Chain, reader: JsonRpcReader): LabelsDetector {
return EthereumLabelsDetector(reader, chain)
}

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
View File

@@ -34,7 +34,6 @@ import io.emeraldpay.etherjar.rpc.json.SyncingJson
import io.emeraldpay.etherjar.rpc.json.TransactionCallJson
import org.slf4j.LoggerFactory
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers
import reactor.kotlin.extra.retry.retryRandomBackoff
@@ -45,10 +44,10 @@ import java.util.concurrent.TimeoutException
open class EthereumUpstreamValidator @JvmOverloads constructor(
private val chain: Chain,
private val upstream: Upstream,
private val options: ChainOptions.Options,
upstream: Upstream,
options: ChainOptions.Options,
private val config: ChainConfig,
) : UpstreamValidator {
) : UpstreamValidator(upstream, options) {
companion object {
private val log = LoggerFactory.getLogger(EthereumUpstreamValidator::class.java)
val scheduler =
@@ -57,7 +56,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
private val objectMapper: ObjectMapper = Global.objectMapper
open fun validate(): Mono<UpstreamAvailability> {
override fun validate(): Mono<UpstreamAvailability> {
return Mono.zip(
validateSyncing(),
validatePeers(),
@@ -127,19 +126,6 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
.onErrorReturn(UpstreamAvailability.UNAVAILABLE)
}
override fun start(): Flux<UpstreamAvailability> {
return Flux.interval(
Duration.ZERO,
Duration.ofSeconds(options.validationInterval.toLong()),
).subscribeOn(scheduler)
.flatMap {
validate()
}
.doOnNext {
log.debug("Status after validation is $it for ${upstream.getId()}")
}
}
override fun validateUpstreamSettings(): Mono<ValidateUpstreamSettingsResult> {
if (options.disableUpstreamValidation) {
return Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_VALID)

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt
View File

@@ -2,9 +2,7 @@ package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.foundation.ChainOptions.Options
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.CachingReader
import io.emeraldpay.dshackle.upstream.EgressSubscription
@@ -16,8 +14,6 @@ import io.emeraldpay.dshackle.upstream.LogsOracle
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.NoIngressSubscription
import io.emeraldpay.dshackle.upstream.NoopCachingReader
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamValidator
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.CallSelector
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
@@ -41,15 +37,6 @@ abstract class AbstractChainSpecific : ChainSpecific {
return { _, _, _ -> NoopCachingReader }
}
override fun validator(
chain: Chain,
upstream: Upstream,
options: Options,
config: ChainConfig,
): UpstreamValidator? {
return null
}
override fun labelDetector(chain: Chain, reader: JsonRpcReader): LabelsDetector? {
return null
}

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

@@ -53,7 +53,7 @@ interface ChainSpecific {
fun makeCachingReaderBuilder(tracer: Tracer): CachingReaderBuilder
fun validator(chain: Chain, upstream: Upstream, options: ChainOptions.Options, config: ChainConfig): UpstreamValidator?
fun validator(chain: Chain, upstream: Upstream, options: ChainOptions.Options, config: ChainConfig): UpstreamValidator
fun labelDetector(chain: Chain, reader: JsonRpcReader): LabelsDetector?

42
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstreamValidator.kt Normal file
View File

@@ -0,0 +1,42 @@
package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.foundation.ChainOptions
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.ValidateUpstreamSettingsResult
import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult.UPSTREAM_VALID
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.util.concurrent.TimeoutException
class GenericUpstreamValidator(
upstream: Upstream,
options: ChainOptions.Options,
private val validator: SingleCallValidator,
) : UpstreamValidator(upstream, options) {
companion object {
private val log = LoggerFactory.getLogger(GenericUpstreamValidator::class.java)
}
override fun validate(): Mono<UpstreamAvailability> {
return upstream.getIngressReader()
.read(validator.method)
.flatMap(JsonRpcResponse::requireResult)
.map { validator.check(it) }
.timeout(
Defaults.timeoutInternal,
Mono.fromCallable { log.warn("No response for ${validator.method.method} from ${upstream.getId()}") }
.then(Mono.error(TimeoutException("Validation timeout for ${validator.method.method}"))),
)
.doOnError { err -> log.error("Error during ${validator.method.method} validation for ${upstream.getId()}", err) }
.onErrorReturn(UpstreamAvailability.UNAVAILABLE)
}
override fun validateUpstreamSettings(): Mono<ValidateUpstreamSettingsResult> {
return Mono.just(UPSTREAM_VALID)
}
}

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

@@ -13,28 +13,30 @@ import io.emeraldpay.dshackle.upstream.CachingReader
import io.emeraldpay.dshackle.upstream.EgressSubscription
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.IngressSubscription
import io.emeraldpay.dshackle.upstream.LabelsDetector
import io.emeraldpay.dshackle.upstream.LogsOracle
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.NoopCachingReader
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.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultPolkadotMethods
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific
import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder
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.rpcclient.JsonRpcRequest
import org.springframework.cloud.sleuth.Tracer
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
import java.math.BigInteger
import java.time.Instant
object PolkadotChainSpecific : AbstractPollChainSpecific() {
private val log = LoggerFactory.getLogger(PolkadotChainSpecific::class.java)
override fun parseBlock(data: ByteArray, upstreamId: String): BlockContainer {
val response = Global.objectMapper.readValue(data, PolkadotBlockResponse::class.java)
@@ -84,21 +86,35 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() {
return { ms -> GenericEgressSubscription(ms, headScheduler, DefaultPolkadotMethods.subs.map { it.first }) }
}
override fun makeCachingReaderBuilder(tracer: Tracer): CachingReaderBuilder {
return { _, _, _ -> NoopCachingReader }
}
override fun validator(
chain: Chain,
upstream: Upstream,
options: Options,
config: ChainConfig,
): UpstreamValidator? {
return null
): UpstreamValidator {
return GenericUpstreamValidator(
upstream,
options,
SingleCallValidator(
JsonRpcRequest("system_health", listOf()),
) { data ->
validate(data, options.minPeers, upstream.getId())
},
)
}
override fun labelDetector(chain: Chain, reader: JsonRpcReader): LabelsDetector? {
return null
fun validate(data: ByteArray, peers: Int, upstreamId: String): UpstreamAvailability {
val resp = Global.objectMapper.readValue(data, PolkadotHealth::class.java)
if (resp.isSyncing) {
log.warn("Polkadot node {} is in syncing state", upstreamId)
return UpstreamAvailability.SYNCING
}
if (resp.shouldHavePeers && resp.peers < peers) {
log.warn("Polkadot node {} should but doesn't have enough peers ({} < {})", upstreamId, resp.peers, peers)
return UpstreamAvailability.IMMATURE
}
return UpstreamAvailability.OK
}
override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription {
@@ -128,3 +144,10 @@ data class PolkadotHeader(
data class PolkadotDigest(
@JsonProperty("logs") var logs: List<String>,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class PolkadotHealth(
@JsonProperty("peers") var peers: Long,
@JsonProperty("isSyncing") var isSyncing: Boolean,
@JsonProperty("shouldHavePeers") var shouldHavePeers: Boolean,
)

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

@@ -4,18 +4,25 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties
import com.fasterxml.jackson.annotation.JsonProperty
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.foundation.ChainOptions.Options
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.DefaultSolanaMethods
import io.emeraldpay.dshackle.upstream.EgressSubscription
import io.emeraldpay.dshackle.upstream.IngressSubscription
import io.emeraldpay.dshackle.upstream.LabelsDetector
import io.emeraldpay.dshackle.upstream.Multistream
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.ethereum.WsSubscriptions
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.rpcclient.JsonRpcRequest
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
@@ -105,6 +112,29 @@ object SolanaChainSpecific : AbstractChainSpecific() {
return JsonRpcRequest("blockUnsubscribe", listOf(subId))
}
override fun validator(
chain: Chain,
upstream: Upstream,
options: Options,
config: ChainConfig,
): UpstreamValidator {
return GenericUpstreamValidator(
upstream,
options,
SingleCallValidator(
JsonRpcRequest("getHealth", listOf()),
) { data ->
val resp = String(data)
if (resp == "\"ok\"") {
UpstreamAvailability.OK
} else {
log.warn("Upstream {} validation failed, solana status is {}", upstream.getId(), resp)
UpstreamAvailability.UNAVAILABLE
}
},
)
}
override fun labelDetector(chain: Chain, reader: JsonRpcReader): LabelsDetector? {
return null
}

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

@@ -2,15 +2,27 @@ package io.emeraldpay.dshackle.upstream.starknet
import com.fasterxml.jackson.annotation.JsonIgnoreProperties
import com.fasterxml.jackson.annotation.JsonProperty
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.foundation.ChainOptions.Options
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.rpcclient.JsonRpcRequest
import org.slf4j.LoggerFactory
import java.math.BigInteger
import java.time.Instant
object StarknetChainSpecific : AbstractPollChainSpecific() {
private val log = LoggerFactory.getLogger(StarknetChainSpecific::class.java)
override fun parseBlock(data: ByteArray, upstreamId: String): BlockContainer {
val block = Global.objectMapper.readValue(data, StarknetBlock::class.java)
@@ -40,6 +52,33 @@ object StarknetChainSpecific : AbstractPollChainSpecific() {
throw NotImplementedError()
}
override fun validator(
chain: Chain,
upstream: Upstream,
options: Options,
config: ChainConfig,
): UpstreamValidator {
return GenericUpstreamValidator(
upstream,
options,
SingleCallValidator(
JsonRpcRequest("starknet_syncing", listOf()),
) { data ->
validate(data, config.laggingLagSize, upstream.getId())
},
)
}
fun validate(data: ByteArray, lagging: Int, upstreamId: String): UpstreamAvailability {
val resp = Global.objectMapper.readValue(data, StarknetSyncing::class.java)
return if (resp.highest - resp.current > lagging) {
log.warn("Starknet node {} is syncing: current={} and highest={}", upstreamId, resp.current, resp.highest)
UpstreamAvailability.SYNCING
} else {
UpstreamAvailability.OK
}
}
override fun latestBlockRequest(): JsonRpcRequest =
JsonRpcRequest("starknet_getBlockWithTxHashes", listOf("latest"))
}
@@ -51,3 +90,9 @@ data class StarknetBlock(
@JsonProperty("timestamp") var timestamp: Instant,
@JsonProperty("parent_hash") var parent: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class StarknetSyncing(
@JsonProperty("current_block_num") var current: Long,
@JsonProperty("highest_block_num") var highest: Long,
)

57
src/test/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecificTest.kt
View File

@@ -1,6 +1,7 @@
package io.emeraldpay.dshackle.upstream.polkadot
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import org.assertj.core.api.Assertions
import org.junit.jupiter.api.Test
@@ -31,6 +32,34 @@ val example = """
}
""".trimIndent()
val healthOk = """{
"peers": 16,
"isSyncing": false,
"shouldHavePeers": true
}
""".trimIndent()
val healthOk2 = """{
"peers": 0,
"isSyncing": false,
"shouldHavePeers": false
}
""".trimIndent()
val healthBadPeers = """{
"peers": 0,
"isSyncing": false,
"shouldHavePeers": true
}
""".trimIndent()
val healthBadSyncing = """{
"peers": 16,
"isSyncing": true,
"shouldHavePeers": true
}
""".trimIndent()
class PolkadotChainSpecificTest {
@Test
fun parseResponse() {
@@ -41,4 +70,32 @@ class PolkadotChainSpecificTest {
Assertions.assertThat(result.upstreamId).isEqualTo("1")
Assertions.assertThat(result.parentHash).isEqualTo(BlockId.from("0xb52a9b51fb698a891cf378b990b0b6a5743e52fa5175b44a8a6d4e0b2cfd0a53"))
}
@Test
fun validateHealth() {
Assertions.assertThat(PolkadotChainSpecific.validate(healthOk.toByteArray(), 10, "test")).isEqualTo(
UpstreamAvailability.OK,
)
}
@Test
fun validateHealthWithoutPeersOk() {
Assertions.assertThat(PolkadotChainSpecific.validate(healthOk2.toByteArray(), 10, "test")).isEqualTo(
UpstreamAvailability.OK,
)
}
@Test
fun validateHealthWithoutPeers() {
Assertions.assertThat(PolkadotChainSpecific.validate(healthBadPeers.toByteArray(), 10, "test")).isEqualTo(
UpstreamAvailability.IMMATURE,
)
}
@Test
fun validateHealthSyncing() {
Assertions.assertThat(PolkadotChainSpecific.validate(healthBadSyncing.toByteArray(), 10, "test")).isEqualTo(
UpstreamAvailability.SYNCING,
)
}
}

33
src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecificTest.kt
View File

@@ -1,6 +1,7 @@
package io.emeraldpay.dshackle.upstream.starknet
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import org.assertj.core.api.Assertions
import org.junit.jupiter.api.Test
@@ -19,6 +20,28 @@ val example = """
}
""".trimIndent()
val syncingGood = """
{
"current_block_hash": "0x2672c3bd6afd698cf979ead19e486cad0c113e1ea7c55467a4c514128b3fe2c",
"current_block_num": 426006,
"highest_block_hash": "0x2672c3bd6afd698cf979ead19e486cad0c113e1ea7c55467a4c514128b3fe2c",
"highest_block_num": 426006,
"starting_block_hash": "0x43734ae66b4f11afa254d0ac5e8f3bc599638c8c92ca97f82e680c4a012005a",
"starting_block_num": 397751
}
""".trimIndent()
val syncingBad = """
{
"current_block_hash": "0x2672c3bd6afd698cf979ead19e486cad0c113e1ea7c55467a4c514128b3fe2c",
"current_block_num": 426006,
"highest_block_hash": "0x2672c3bd6afd698cf979ead19e486cad0c113e1ea7c55467a4c514128b3fe2c",
"highest_block_num": 526006,
"starting_block_hash": "0x43734ae66b4f11afa254d0ac5e8f3bc599638c8c92ca97f82e680c4a012005a",
"starting_block_num": 397751
}
""".trimIndent()
class StarknetChainSpecificTest {
@Test
fun parseResponse() {
@@ -29,4 +52,14 @@ class StarknetChainSpecificTest {
Assertions.assertThat(result.upstreamId).isEqualTo("1")
Assertions.assertThat(result.parentHash).isEqualTo(BlockId.from("07cc1e178c848b0bdfa047a414d9a8bee4f6cb76f25f312f2d42e058ea91d78b"))
}
@Test
fun validateOk() {
Assertions.assertThat(StarknetChainSpecific.validate(syncingGood.toByteArray(), 10, "test")).isEqualTo(UpstreamAvailability.OK)
}
@Test
fun validateSyncing() {
Assertions.assertThat(StarknetChainSpecific.validate(syncingBad.toByteArray(), 10, "test")).isEqualTo(UpstreamAvailability.SYNCING)
}
}