Checking eth_call return limit during node validation, currently 200k (#246)
Added eth_call return limit check to node validation. Currently limit is 200k
This commit is contained in:
@@ -11,6 +11,7 @@ data class ChainsConfig(private val chains: Map<Chain, RawChainConfig>, val curr
|
||||
data class RawChainConfig(
|
||||
var syncingLagSize: Int? = null,
|
||||
var laggingLagSize: Int? = null,
|
||||
var callLimitContract: String? = null,
|
||||
var options: UpstreamsConfig.PartialOptions? = null
|
||||
) {
|
||||
|
||||
@@ -26,11 +27,12 @@ data class ChainsConfig(private val chains: Map<Chain, RawChainConfig>, val curr
|
||||
data class ChainConfig(
|
||||
val syncingLagSize: Int,
|
||||
val laggingLagSize: Int,
|
||||
val options: UpstreamsConfig.PartialOptions
|
||||
val options: UpstreamsConfig.PartialOptions,
|
||||
val callLimitContract: String?
|
||||
) {
|
||||
companion object {
|
||||
@JvmStatic
|
||||
fun default() = ChainConfig(6, 1, UpstreamsConfig.PartialOptions())
|
||||
fun default() = ChainConfig(6, 1, UpstreamsConfig.PartialOptions(), null)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -42,7 +44,8 @@ data class ChainsConfig(private val chains: Map<Chain, RawChainConfig>, val curr
|
||||
return ChainConfig(
|
||||
laggingLagSize = raw.laggingLagSize ?: default.laggingLagSize ?: panic(),
|
||||
syncingLagSize = raw.syncingLagSize ?: default.syncingLagSize ?: panic(),
|
||||
options = options
|
||||
options = options,
|
||||
callLimitContract = raw.callLimitContract
|
||||
)
|
||||
}
|
||||
|
||||
@@ -57,7 +60,8 @@ data class ChainsConfig(private val chains: Map<Chain, RawChainConfig>, val curr
|
||||
) = RawChainConfig(
|
||||
syncingLagSize = patch?.syncingLagSize ?: current.syncingLagSize,
|
||||
laggingLagSize = patch?.laggingLagSize ?: current.laggingLagSize,
|
||||
options = patch?.options ?: current.options
|
||||
options = patch?.options ?: current.options,
|
||||
callLimitContract = patch?.callLimitContract ?: current.callLimitContract
|
||||
)
|
||||
|
||||
private fun merge(
|
||||
|
||||
@@ -53,6 +53,9 @@ class ChainsConfigReader(
|
||||
rawConfig.laggingLagSize = it
|
||||
}
|
||||
}
|
||||
getValueAsString(node, "call-validate-contract")?.let {
|
||||
rawConfig.callLimitContract = it
|
||||
}
|
||||
upstreamsConfigReader.tryReadOptions(node)?.let {
|
||||
rawConfig.options = it
|
||||
}
|
||||
|
||||
@@ -36,7 +36,8 @@ open class UpstreamsConfig {
|
||||
val providesBalance: Boolean?,
|
||||
val validatePeers: Boolean,
|
||||
val minPeers: Int,
|
||||
val validateSyncing: Boolean
|
||||
val validateSyncing: Boolean,
|
||||
val validateCallLimit: Boolean
|
||||
)
|
||||
|
||||
open class PartialOptions {
|
||||
@@ -51,6 +52,7 @@ open class UpstreamsConfig {
|
||||
var timeout: Duration? = null
|
||||
var providesBalance: Boolean? = null
|
||||
var validatePeers: Boolean? = null
|
||||
var validateCalllimit: Boolean? = null
|
||||
var minPeers: Int? = null
|
||||
set(value) {
|
||||
require(value == null || value >= 0) {
|
||||
@@ -71,6 +73,7 @@ open class UpstreamsConfig {
|
||||
copy.validationInterval = firstNonNull(overwrites.validationInterval, this.validationInterval)
|
||||
copy.providesBalance = firstNonNull(overwrites.providesBalance, this.providesBalance)
|
||||
copy.validateSyncing = firstNonNull(overwrites.validateSyncing, this.validateSyncing)
|
||||
copy.validateCalllimit = firstNonNull(overwrites.validateCalllimit, this.validateCalllimit)
|
||||
copy.timeout = firstNonNull(overwrites.timeout, this.timeout)
|
||||
return copy
|
||||
}
|
||||
@@ -83,7 +86,8 @@ open class UpstreamsConfig {
|
||||
this.providesBalance,
|
||||
firstNonNull(this.validatePeers, true)!!,
|
||||
firstNonNull(this.minPeers, 1)!!,
|
||||
firstNonNull(this.validateSyncing, true)!!
|
||||
firstNonNull(this.validateSyncing, true)!!,
|
||||
firstNonNull(this.validateCalllimit, true)!!
|
||||
)
|
||||
|
||||
companion object {
|
||||
|
||||
@@ -353,6 +353,9 @@ class UpstreamsConfigReader(
|
||||
getValueAsBool(values, "validate-syncing")?.let {
|
||||
options.validateSyncing = it
|
||||
}
|
||||
getValueAsBool(values, "validate-call-limit")?.let {
|
||||
options.validateCalllimit = it
|
||||
}
|
||||
getValueAsInt(values, "min-peers")?.let {
|
||||
options.minPeers = it
|
||||
}
|
||||
|
||||
@@ -45,7 +45,7 @@ open class EthereumLikeRpcUpstream(
|
||||
chainConfig: ChainsConfig.ChainConfig,
|
||||
skipEnhance: Boolean
|
||||
) : EthereumLikeUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled {
|
||||
private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
|
||||
private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions(), chainConfig.callLimitContract)
|
||||
private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, skipEnhance)
|
||||
private val labelsDetector = EthereumLabelsDetector(this.getIngressReader())
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ abstract class EthereumLikeUpstream(
|
||||
role: UpstreamsConfig.UpstreamRole,
|
||||
targets: CallMethods?,
|
||||
private val node: QuorumForLabels.QuorumItem?,
|
||||
chainConfig: ChainsConfig.ChainConfig
|
||||
val chainConfig: ChainsConfig.ChainConfig
|
||||
) : DefaultUpstream(id, hash, options, role, targets, node, chainConfig) {
|
||||
|
||||
private val capabilities = if (options.providesBalance != false) {
|
||||
|
||||
@@ -24,21 +24,26 @@ import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
|
||||
import io.emeraldpay.etherjar.domain.Address
|
||||
import io.emeraldpay.etherjar.hex.HexData
|
||||
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.util.function.Tuple2
|
||||
import reactor.util.function.Tuple3
|
||||
import java.time.Duration
|
||||
import java.util.concurrent.Executors
|
||||
import java.util.concurrent.TimeoutException
|
||||
|
||||
open class EthereumUpstreamValidator(
|
||||
open class EthereumUpstreamValidator @JvmOverloads constructor(
|
||||
private val upstream: Upstream,
|
||||
private val options: UpstreamsConfig.Options
|
||||
private val options: UpstreamsConfig.Options,
|
||||
private val callLimitContract: String? = null
|
||||
) {
|
||||
private var callLimitSucceed: Boolean = false
|
||||
companion object {
|
||||
private val log = LoggerFactory.getLogger(EthereumUpstreamValidator::class.java)
|
||||
val scheduler =
|
||||
@@ -50,15 +55,54 @@ open class EthereumUpstreamValidator(
|
||||
open fun validate(): Mono<UpstreamAvailability> {
|
||||
return Mono.zip(
|
||||
validateSyncing(),
|
||||
validatePeers()
|
||||
validatePeers(),
|
||||
validateCallLimit()
|
||||
)
|
||||
.map(::resolve)
|
||||
.defaultIfEmpty(UpstreamAvailability.UNAVAILABLE)
|
||||
.onErrorReturn(UpstreamAvailability.UNAVAILABLE)
|
||||
}
|
||||
|
||||
fun resolve(results: Tuple2<UpstreamAvailability, UpstreamAvailability>): UpstreamAvailability {
|
||||
return if (results.t1.isBetterTo(results.t2)) results.t2 else results.t1
|
||||
fun resolve(results: Tuple3<UpstreamAvailability, UpstreamAvailability, UpstreamAvailability>): UpstreamAvailability {
|
||||
val cp = Comparator { avail1: UpstreamAvailability, avail2: UpstreamAvailability -> if (avail1.isBetterTo(avail2)) -1 else 1 }
|
||||
return listOf(results.t1, results.t2, results.t3).sortedWith(cp).last()
|
||||
}
|
||||
|
||||
fun validateCallLimit(): Mono<UpstreamAvailability> {
|
||||
// do not rerun this check after first success because it's more expensive than others
|
||||
if (!options.validateCallLimit || callLimitContract == null || callLimitSucceed) {
|
||||
return Mono.just(UpstreamAvailability.OK)
|
||||
}
|
||||
return upstream.getIngressReader()
|
||||
.read(
|
||||
JsonRpcRequest(
|
||||
"eth_call",
|
||||
listOf(
|
||||
TransactionCallJson(
|
||||
Address.from(callLimitContract),
|
||||
// calling contract with param 200_000, meaning it will generate 200k symbols or response
|
||||
// 30d40 — 200_000
|
||||
HexData.from("0xd8a26e3a0000000000000000000000000000000000000000000000000000000000030d40")
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
.flatMap(JsonRpcResponse::requireResult)
|
||||
.doOnError {
|
||||
log.error(
|
||||
"Node ${upstream.getId()} is incorrectly configured. " +
|
||||
"You need to set up your return limit to at least 200000." +
|
||||
"Erigon config example: https://github.com/ledgerwatch/erigon/blob/devel/cmd/utils/flags.go#L364. "
|
||||
)
|
||||
}
|
||||
.map { UpstreamAvailability.OK }
|
||||
.timeout(
|
||||
Defaults.timeoutInternal,
|
||||
Mono.fromCallable { log.error("No response for eth_call limit check from ${upstream.getId()}") }
|
||||
.then(Mono.error(TimeoutException("Validation timeout for call limit")))
|
||||
)
|
||||
.doOnSuccess { callLimitSucceed = true }
|
||||
.onErrorReturn(UpstreamAvailability.UNAVAILABLE)
|
||||
}
|
||||
|
||||
fun validateSyncing(): Mono<UpstreamAvailability> {
|
||||
|
||||
@@ -7,6 +7,12 @@ chain-settings:
|
||||
lagging: 1
|
||||
chains:
|
||||
- id: eth
|
||||
call-validate-contract: 0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96
|
||||
lags:
|
||||
syncing: 6
|
||||
lagging: 1
|
||||
- id: goerli
|
||||
call-validate-contract: 0xCD9303A1F6da2a68f465A579a24cc2Ee5AE2192f
|
||||
lags:
|
||||
syncing: 6
|
||||
lagging: 1
|
||||
|
||||
@@ -38,6 +38,7 @@ class ChainsConfigReaderSpec extends Specification {
|
||||
then:
|
||||
eth.laggingLagSize == 1
|
||||
eth.syncingLagSize == 6
|
||||
eth.callLimitContract == "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96"
|
||||
|
||||
pol.laggingLagSize == 10
|
||||
pol.syncingLagSize == 20
|
||||
@@ -48,8 +49,5 @@ class ChainsConfigReaderSpec extends Specification {
|
||||
|
||||
sep.laggingLagSize == 1
|
||||
sep.syncingLagSize == 10
|
||||
|
||||
eth.laggingLagSize == 1
|
||||
eth.syncingLagSize == 6
|
||||
}
|
||||
}
|
||||
|
||||
@@ -450,11 +450,13 @@ class UpstreamsConfigReaderSpec extends Specification {
|
||||
disableValidation == false
|
||||
validateSyncing == true
|
||||
validatePeers == false
|
||||
validateCalllimit == true
|
||||
}
|
||||
with(act.upstreams.get(1).options) {
|
||||
disableValidation == false
|
||||
validateSyncing == false
|
||||
validatePeers == false
|
||||
validateCalllimit == false
|
||||
}
|
||||
with(act.upstreams.get(2).options) {
|
||||
disableValidation == true
|
||||
@@ -633,7 +635,7 @@ class UpstreamsConfigReaderSpec extends Specification {
|
||||
def options = partialOptions.buildOptions()
|
||||
then:
|
||||
options == new UpstreamsConfig.Options(
|
||||
false, 30, Duration.ofSeconds(60), null, true, 1, true
|
||||
false, 30, Duration.ofSeconds(60), null, true, 1, true, true
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.test
|
||||
|
||||
import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig.Options
|
||||
import io.emeraldpay.dshackle.data.BlockContainer
|
||||
@@ -70,7 +71,7 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream {
|
||||
methods,
|
||||
new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)),
|
||||
new ConnectorFactoryMock(api, new EthereumHeadMock()),
|
||||
ChainsConfig.ChainConfig.default(),
|
||||
ChainConfig.default(),
|
||||
true
|
||||
)
|
||||
this.ethereumHeadMock = this.getHead() as EthereumHeadMock
|
||||
|
||||
@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.FileResolver
|
||||
import io.emeraldpay.dshackle.cache.Caches
|
||||
import io.emeraldpay.dshackle.cache.CachesFactory
|
||||
import io.emeraldpay.dshackle.config.CacheConfig
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
|
||||
import io.emeraldpay.dshackle.data.BlockContainer
|
||||
import io.emeraldpay.dshackle.data.BlockId
|
||||
import io.emeraldpay.dshackle.reader.EmptyReader
|
||||
|
||||
@@ -15,13 +15,17 @@
|
||||
*/
|
||||
package io.emeraldpay.dshackle.upstream.ethereum
|
||||
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.reader.Reader
|
||||
import io.emeraldpay.dshackle.test.ApiReaderMock
|
||||
import io.emeraldpay.dshackle.test.TestingCommons
|
||||
import io.emeraldpay.dshackle.upstream.Head
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
|
||||
import io.emeraldpay.etherjar.hex.HexData
|
||||
import io.emeraldpay.etherjar.rpc.RpcResponseError
|
||||
import io.emeraldpay.etherjar.rpc.json.TransactionCallJson
|
||||
import io.emeraldpay.etherjar.domain.Address
|
||||
import reactor.core.publisher.Mono
|
||||
import reactor.util.function.Tuples
|
||||
import spock.lang.Specification
|
||||
@@ -36,18 +40,18 @@ class EthereumUpstreamValidatorSpec extends Specification {
|
||||
setup:
|
||||
def validator = new EthereumUpstreamValidator(Stub(EthereumLikeUpstream), UpstreamsConfig.PartialOptions.getDefaults().buildOptions())
|
||||
expect:
|
||||
validator.resolve(Tuples.of(sync, peers)) == exp
|
||||
validator.resolve(Tuples.of(sync, peers, call)) == exp
|
||||
where:
|
||||
exp | sync | peers
|
||||
OK | OK | OK
|
||||
IMMATURE | OK | IMMATURE
|
||||
UNAVAILABLE | OK | UNAVAILABLE
|
||||
SYNCING | SYNCING | OK
|
||||
SYNCING | SYNCING | IMMATURE
|
||||
UNAVAILABLE | SYNCING | UNAVAILABLE
|
||||
UNAVAILABLE | UNAVAILABLE | OK
|
||||
UNAVAILABLE | UNAVAILABLE | IMMATURE
|
||||
UNAVAILABLE | UNAVAILABLE | UNAVAILABLE
|
||||
exp | sync | peers | call
|
||||
OK | OK | OK | OK
|
||||
IMMATURE | OK | IMMATURE | OK
|
||||
UNAVAILABLE | OK | UNAVAILABLE | OK
|
||||
SYNCING | SYNCING | OK | OK
|
||||
SYNCING | SYNCING | IMMATURE | OK
|
||||
UNAVAILABLE | SYNCING | UNAVAILABLE | OK
|
||||
UNAVAILABLE | UNAVAILABLE | OK | OK
|
||||
UNAVAILABLE | UNAVAILABLE | IMMATURE | OK
|
||||
UNAVAILABLE | UNAVAILABLE | UNAVAILABLE | OK
|
||||
}
|
||||
|
||||
def "Doesnt check eth_syncing when disabled"() {
|
||||
@@ -62,7 +66,7 @@ class EthereumUpstreamValidatorSpec extends Specification {
|
||||
def act = validator.validateSyncing().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act == OK
|
||||
0 * up.getApi()
|
||||
0 * up.getIngressReader()
|
||||
}
|
||||
|
||||
def "Syncing is OK when false returned from upstream"() {
|
||||
@@ -175,7 +179,7 @@ class EthereumUpstreamValidatorSpec extends Specification {
|
||||
def act = validator.validatePeers().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act == OK
|
||||
0 * up.getApi()
|
||||
0 * up.getIngressReader()
|
||||
}
|
||||
|
||||
def "Peers is IMMATURE when state returned too few peers"() {
|
||||
@@ -253,4 +257,66 @@ class EthereumUpstreamValidatorSpec extends Specification {
|
||||
then:
|
||||
act == UNAVAILABLE
|
||||
}
|
||||
|
||||
def "Doesnt check call limit when disabled"() {
|
||||
setup:
|
||||
def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
|
||||
it.validateCalllimit = false
|
||||
}.buildOptions()
|
||||
def up = Mock(EthereumLikeUpstream)
|
||||
def validator = new EthereumUpstreamValidator(up, options)
|
||||
|
||||
when:
|
||||
def act = validator.validateCallLimit().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act == OK
|
||||
0 * up.getIngressReader()
|
||||
}
|
||||
|
||||
def "Upstream available if not error from call limit check"() {
|
||||
setup:
|
||||
def options = UpstreamsConfig.PartialOptions.getDefaults().buildOptions()
|
||||
def up = TestingCommons.upstream(
|
||||
new ApiReaderMock().tap {
|
||||
answerOnce("eth_call", [new TransactionCallJson(
|
||||
Address.from("0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96"),
|
||||
HexData.from("0xd8a26e3a0000000000000000000000000000000000000000000000000000000000030d40")
|
||||
)], "0x00000000000000000000")
|
||||
}
|
||||
)
|
||||
def validator = new EthereumUpstreamValidator(up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
|
||||
|
||||
when:
|
||||
def act = validator.validateCallLimit().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act == OK
|
||||
when:
|
||||
def act2 = validator.validateCallLimit().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act2 == OK
|
||||
}
|
||||
|
||||
def "Upstream not available if error returned on call limit check"() {
|
||||
setup:
|
||||
def options = UpstreamsConfig.PartialOptions.getDefaults().buildOptions()
|
||||
def up = TestingCommons.upstream(
|
||||
new ApiReaderMock().tap {
|
||||
answer("eth_call", [new TransactionCallJson(
|
||||
Address.from("0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96"),
|
||||
HexData.from("0xd8a26e3a0000000000000000000000000000000000000000000000000000000000030d40")
|
||||
)], new RpcResponseError(RpcResponseError.CODE_INVALID_REQUEST, "Too long"))
|
||||
}
|
||||
)
|
||||
def validator = new EthereumUpstreamValidator(up, options, "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96")
|
||||
|
||||
when:
|
||||
def act = validator.validateCallLimit().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act == UNAVAILABLE
|
||||
when:
|
||||
def act2 = validator.validateCallLimit().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act2 == UNAVAILABLE
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ chain-settings:
|
||||
lagging: 1
|
||||
chains:
|
||||
- id: eth
|
||||
call-validate-contract: 0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96
|
||||
lags:
|
||||
syncing: 6
|
||||
lagging: 1
|
||||
|
||||
@@ -8,6 +8,7 @@ upstreams:
|
||||
disable-validation: false
|
||||
validate-syncing: true
|
||||
validate-peers: false
|
||||
validate-call-limit: true
|
||||
connection:
|
||||
ethereum:
|
||||
rpc:
|
||||
@@ -19,6 +20,7 @@ upstreams:
|
||||
disable-validation: false
|
||||
validate-syncing: false
|
||||
validate-peers: false
|
||||
validate-call-limit: false
|
||||
connection:
|
||||
ethereum:
|
||||
rpc:
|
||||
|
||||
Reference in New Issue
Block a user