Autodetect methods (#562)
* Autodetect methods * add polkadot magic method detection * fix lint * change config.methods writing * fix log * change Flux to Mono.zip * change block to subscribe * split chain method detectors * eth method detector use rpc_modules * rm UpstreamRpcModulesDetector.kt * add tests * getAllMethods for eth * prefer enable/disable methods from config * update tests * whiter list & full is not available error check * prefer config method group * mv notAvailableRegexps to class variable
This commit is contained in:
@@ -84,7 +84,7 @@ open class GenericUpstreamCreator(
|
||||
connectorFactory,
|
||||
cs::validator,
|
||||
cs::upstreamSettingsDetector,
|
||||
cs::upstreamRpcModulesDetector,
|
||||
cs::upstreamRpcMethodsDetector,
|
||||
buildMethodsFun,
|
||||
cs::lowerBoundService,
|
||||
cs::finalizationDetectorBuilder,
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
package io.emeraldpay.dshackle.upstream
|
||||
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.CallParams
|
||||
import org.slf4j.Logger
|
||||
import org.slf4j.LoggerFactory
|
||||
import reactor.core.publisher.Mono
|
||||
|
||||
typealias UpstreamRpcMethodsDetectorBuilder = (Upstream, UpstreamsConfig.Upstream<*>?) -> UpstreamRpcMethodsDetector?
|
||||
|
||||
abstract class UpstreamRpcMethodsDetector(
|
||||
private val upstream: Upstream,
|
||||
private val config: UpstreamsConfig.Upstream<*>? = null,
|
||||
) {
|
||||
protected val log: Logger = LoggerFactory.getLogger(this::class.java)
|
||||
|
||||
private val notAvailableRegexps =
|
||||
listOf(
|
||||
"method ([A-Za-z0-9_]+) does not exist/is not available",
|
||||
"([A-Za-z0-9_]+) found but the containing module is disabled",
|
||||
"Method not found",
|
||||
"The method ([A-Za-z0-9_]+) is not available",
|
||||
).map { s -> s.toRegex() }
|
||||
|
||||
open fun detectRpcMethods(): Mono<Map<String, Boolean>> = detectByMagicMethod().switchIfEmpty(detectByMethod())
|
||||
|
||||
protected fun detectByMethod(): Mono<Map<String, Boolean>> =
|
||||
Mono.zip(
|
||||
rpcMethods().map {
|
||||
Mono
|
||||
.just(it)
|
||||
.flatMap { (method, param) ->
|
||||
upstream
|
||||
.getIngressReader()
|
||||
.read(ChainRequest(method, param))
|
||||
.flatMap(ChainResponse::requireResult)
|
||||
.map { method to true }
|
||||
.onErrorResume { err ->
|
||||
val notAvailableError =
|
||||
notAvailableRegexps.any { s -> s.containsMatchIn(err.message ?: "") }
|
||||
if (notAvailableError) {
|
||||
Mono.just(method to false)
|
||||
} else {
|
||||
Mono.empty()
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
) {
|
||||
it
|
||||
.map { p -> p as Pair<String, Boolean> }
|
||||
.associate { (method, enabled) -> method to enabled }
|
||||
}
|
||||
|
||||
protected abstract fun detectByMagicMethod(): Mono<Map<String, Boolean>>
|
||||
|
||||
protected abstract fun rpcMethods(): Set<Pair<String, CallParams>>
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
package io.emeraldpay.dshackle.upstream
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||
import org.slf4j.Logger
|
||||
import org.slf4j.LoggerFactory
|
||||
import reactor.core.publisher.Mono
|
||||
|
||||
typealias UpstreamRpcModulesDetectorBuilder = (Upstream) -> UpstreamRpcModulesDetector?
|
||||
|
||||
abstract class UpstreamRpcModulesDetector(
|
||||
private val upstream: Upstream,
|
||||
) {
|
||||
protected val log: Logger = LoggerFactory.getLogger(this::class.java)
|
||||
|
||||
open fun detectRpcModules(): Mono<HashMap<String, String>> {
|
||||
return upstream.getIngressReader()
|
||||
.read(rpcModulesRequest())
|
||||
.flatMap(ChainResponse::requireResult)
|
||||
.map(::parseRpcModules)
|
||||
.onErrorResume {
|
||||
log.warn("Can't detect rpc_modules of upstream ${upstream.getId()}, reason - {}", it.message)
|
||||
Mono.just(HashMap())
|
||||
}
|
||||
}
|
||||
|
||||
protected abstract fun rpcModulesRequest(): ChainRequest
|
||||
|
||||
protected abstract fun parseRpcModules(data: ByteArray): HashMap<String, String>
|
||||
}
|
||||
|
||||
class BasicEthUpstreamRpcModulesDetector(
|
||||
upstream: Upstream,
|
||||
) : UpstreamRpcModulesDetector(upstream) {
|
||||
override fun rpcModulesRequest(): ChainRequest = ChainRequest("rpc_modules", ListParams())
|
||||
|
||||
override fun parseRpcModules(data: ByteArray): HashMap<String, String> {
|
||||
return Global.objectMapper.readValue(data, object : TypeReference<HashMap<String, String>>() {})
|
||||
}
|
||||
}
|
||||
@@ -809,4 +809,12 @@ class DefaultEthereumMethods(
|
||||
override fun getSupportedMethods(): Set<String> {
|
||||
return allowedMethods.plus(hardcodedMethods).toSortedSet()
|
||||
}
|
||||
|
||||
fun getAllMethods(): Set<String> =
|
||||
getSupportedMethods()
|
||||
.plus(getGroupMethods("filter"))
|
||||
.plus(getGroupMethods("trace"))
|
||||
.plus(getGroupMethods("debug"))
|
||||
.plus(getChainSpecificMethods(chain))
|
||||
.toSet()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
package io.emeraldpay.dshackle.upstream.ethereum
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||
import io.emeraldpay.dshackle.upstream.ChainResponse
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetector
|
||||
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.CallParams
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||
import reactor.core.publisher.Mono
|
||||
|
||||
class BasicEthUpstreamRpcMethodsDetector(
|
||||
private val upstream: Upstream,
|
||||
private val config: UpstreamsConfig.Upstream<*>,
|
||||
) : UpstreamRpcMethodsDetector(upstream) {
|
||||
override fun detectByMagicMethod(): Mono<Map<String, Boolean>> =
|
||||
upstream
|
||||
.getIngressReader()
|
||||
.read(ChainRequest("rpc_modules", ListParams()))
|
||||
.flatMap(ChainResponse::requireResult)
|
||||
.map(::parseRpcModules)
|
||||
// force check all methods from rpcMethods
|
||||
.zipWith(detectByMethod()) { a, b ->
|
||||
a.plus(b)
|
||||
}.onErrorResume {
|
||||
log.warn("Can't detect rpc_modules of upstream ${upstream.getId()}, reason - {}", it.message)
|
||||
Mono.empty()
|
||||
}
|
||||
|
||||
override fun rpcMethods(): Set<Pair<String, CallParams>> =
|
||||
setOf(
|
||||
"eth_getBlockReceipts" to ListParams("latest"),
|
||||
)
|
||||
|
||||
private fun parseRpcModules(data: ByteArray): Map<String, Boolean> {
|
||||
val modules = Global.objectMapper.readValue(data, object : TypeReference<HashMap<String, String>>() {})
|
||||
return DefaultEthereumMethods(upstream.getChain())
|
||||
.getAllMethods()
|
||||
.associateWith { method ->
|
||||
if (config.methodGroups?.enabled?.any { group -> method.startsWith(group) } == true) {
|
||||
return@associateWith true
|
||||
}
|
||||
if (config.methodGroups?.disabled?.any { group -> method.startsWith(group) } == true) {
|
||||
return@associateWith false
|
||||
}
|
||||
modules.any { (module, _) -> method.startsWith(module) }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4,10 +4,10 @@ import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.cache.Caches
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.data.BlockContainer
|
||||
import io.emeraldpay.dshackle.foundation.ChainOptions.Options
|
||||
import io.emeraldpay.dshackle.reader.ChainReader
|
||||
import io.emeraldpay.dshackle.upstream.BasicEthUpstreamRpcModulesDetector
|
||||
import io.emeraldpay.dshackle.upstream.CachingReader
|
||||
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||
import io.emeraldpay.dshackle.upstream.EgressSubscription
|
||||
@@ -19,7 +19,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
|
||||
import io.emeraldpay.dshackle.upstream.SingleValidator
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcModulesDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector
|
||||
import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
|
||||
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
||||
@@ -180,9 +180,10 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
|
||||
return ChainIdValidator(upstream, chain, reader)
|
||||
}
|
||||
|
||||
override fun upstreamRpcModulesDetector(upstream: Upstream): UpstreamRpcModulesDetector {
|
||||
return BasicEthUpstreamRpcModulesDetector(upstream)
|
||||
}
|
||||
override fun upstreamRpcMethodsDetector(
|
||||
upstream: Upstream,
|
||||
config: UpstreamsConfig.Upstream<*>?,
|
||||
): UpstreamRpcMethodsDetector? = config?.let { BasicEthUpstreamRpcMethodsDetector(upstream, it) }
|
||||
|
||||
override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService {
|
||||
return EthereumLowerBoundService(chain, upstream)
|
||||
|
||||
@@ -3,6 +3,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.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.config.hot.CompatibleVersionsRules
|
||||
import io.emeraldpay.dshackle.data.BlockContainer
|
||||
import io.emeraldpay.dshackle.foundation.ChainOptions.Options
|
||||
@@ -20,7 +21,7 @@ import io.emeraldpay.dshackle.upstream.NoopCachingReader
|
||||
import io.emeraldpay.dshackle.upstream.SingleValidator
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcModulesDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamValidator
|
||||
import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
|
||||
@@ -68,9 +69,10 @@ abstract class AbstractChainSpecific : ChainSpecific {
|
||||
return null
|
||||
}
|
||||
|
||||
override fun upstreamRpcModulesDetector(upstream: Upstream): UpstreamRpcModulesDetector? {
|
||||
return null
|
||||
}
|
||||
override fun upstreamRpcMethodsDetector(
|
||||
upstream: Upstream,
|
||||
config: UpstreamsConfig.Upstream<*>?,
|
||||
): UpstreamRpcMethodsDetector? = null
|
||||
|
||||
override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription {
|
||||
return NoIngressSubscription()
|
||||
|
||||
@@ -12,6 +12,7 @@ import io.emeraldpay.dshackle.BlockchainType.UNKNOWN
|
||||
import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.cache.Caches
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.config.hot.CompatibleVersionsRules
|
||||
import io.emeraldpay.dshackle.data.BlockContainer
|
||||
import io.emeraldpay.dshackle.foundation.ChainOptions
|
||||
@@ -25,7 +26,7 @@ import io.emeraldpay.dshackle.upstream.LogsOracle
|
||||
import io.emeraldpay.dshackle.upstream.Multistream
|
||||
import io.emeraldpay.dshackle.upstream.SingleValidator
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcModulesDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamValidator
|
||||
import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
|
||||
@@ -86,7 +87,10 @@ interface ChainSpecific {
|
||||
|
||||
fun chainSettingsValidator(chain: Chain, upstream: Upstream, reader: ChainReader?): SingleValidator<ValidateUpstreamSettingsResult>?
|
||||
|
||||
fun upstreamRpcModulesDetector(upstream: Upstream): UpstreamRpcModulesDetector?
|
||||
fun upstreamRpcMethodsDetector(
|
||||
upstream: Upstream,
|
||||
config: UpstreamsConfig.Upstream<*>?,
|
||||
): UpstreamRpcMethodsDetector?
|
||||
|
||||
fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription
|
||||
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package io.emeraldpay.dshackle.upstream.generic
|
||||
|
||||
import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.Defaults
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig.Labels
|
||||
@@ -18,8 +17,8 @@ import io.emeraldpay.dshackle.upstream.IngressSubscription
|
||||
import io.emeraldpay.dshackle.upstream.UNKNOWN_CLIENT_VERSION
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcModulesDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcModulesDetectorBuilder
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetector
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetectorBuilder
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetectorBuilder
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamValidator
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamValidatorBuilder
|
||||
@@ -62,8 +61,8 @@ open class GenericUpstream(
|
||||
lowerBoundServiceBuilder: LowerBoundServiceBuilder,
|
||||
finalizationDetectorBuilder: FinalizationDetectorBuilder,
|
||||
versionRules: Supplier<CompatibleVersionsRules?>,
|
||||
) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig, chain), Lifecycle {
|
||||
|
||||
) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig, chain),
|
||||
Lifecycle {
|
||||
constructor(
|
||||
config: UpstreamsConfig.Upstream<*>,
|
||||
chain: Chain,
|
||||
@@ -74,14 +73,29 @@ open class GenericUpstream(
|
||||
connectorFactory: ConnectorFactory,
|
||||
validatorBuilder: UpstreamValidatorBuilder,
|
||||
upstreamSettingsDetectorBuilder: UpstreamSettingsDetectorBuilder,
|
||||
upstreamRpcModulesDetectorBuilder: UpstreamRpcModulesDetectorBuilder,
|
||||
upstreamRpcMethodsDetectorBuilder: UpstreamRpcMethodsDetectorBuilder,
|
||||
buildMethods: (UpstreamsConfig.Upstream<*>, Chain) -> CallMethods,
|
||||
lowerBoundServiceBuilder: LowerBoundServiceBuilder,
|
||||
finalizationDetectorBuilder: FinalizationDetectorBuilder,
|
||||
versionRules: Supplier<CompatibleVersionsRules?>,
|
||||
) : this(config.id!!, chain, hash, options, config.role, buildMethods(config, chain), node, chainConfig, connectorFactory, validatorBuilder, upstreamSettingsDetectorBuilder, lowerBoundServiceBuilder, finalizationDetectorBuilder, versionRules) {
|
||||
rpcModulesDetector = upstreamRpcModulesDetectorBuilder(this)
|
||||
detectRpcModules(config, buildMethods)
|
||||
) : this(
|
||||
config.id!!,
|
||||
chain,
|
||||
hash,
|
||||
options,
|
||||
config.role,
|
||||
buildMethods(config, chain),
|
||||
node,
|
||||
chainConfig,
|
||||
connectorFactory,
|
||||
validatorBuilder,
|
||||
upstreamSettingsDetectorBuilder,
|
||||
lowerBoundServiceBuilder,
|
||||
finalizationDetectorBuilder,
|
||||
versionRules,
|
||||
) {
|
||||
rpcMethodsDetector = upstreamRpcMethodsDetectorBuilder(this, config)
|
||||
detectRpcMethods(config, buildMethods)
|
||||
}
|
||||
|
||||
private val validator: UpstreamValidator? = validatorBuilder(chain, this, getOptions(), chainConfig, versionRules)
|
||||
@@ -93,7 +107,7 @@ open class GenericUpstream(
|
||||
protected val connector: GenericConnector = connectorFactory.create(this, chain)
|
||||
private var livenessSubscription: Disposable? = null
|
||||
private val settingsDetector = upstreamSettingsDetectorBuilder(chain, this)
|
||||
private var rpcModulesDetector: UpstreamRpcModulesDetector? = null
|
||||
private var rpcMethodsDetector: UpstreamRpcMethodsDetector? = null
|
||||
|
||||
private val lowerBoundService = lowerBoundServiceBuilder(chain, this)
|
||||
|
||||
@@ -168,10 +182,12 @@ open class GenericUpstream(
|
||||
connector.stop()
|
||||
return
|
||||
}
|
||||
|
||||
UPSTREAM_SETTINGS_ERROR -> {
|
||||
log.warn("Non fatal upstream settings error, continue validation...")
|
||||
connector.getHead().stop()
|
||||
}
|
||||
|
||||
UPSTREAM_VALID -> {
|
||||
isUpstreamValid.set(true)
|
||||
upstreamStart()
|
||||
@@ -247,31 +263,44 @@ open class GenericUpstream(
|
||||
}.subscribe()
|
||||
}
|
||||
|
||||
private fun detectRpcModules(config: UpstreamsConfig.Upstream<*>, buildMethods: (UpstreamsConfig.Upstream<*>, Chain) -> CallMethods) {
|
||||
private fun detectRpcMethods(
|
||||
config: UpstreamsConfig.Upstream<*>,
|
||||
buildMethods: (UpstreamsConfig.Upstream<*>, Chain) -> CallMethods,
|
||||
) {
|
||||
try {
|
||||
val rpcDetector = rpcModulesDetector?.detectRpcModules()?.block(Defaults.internalCallsTimeout)
|
||||
?: HashMap<String, String>()
|
||||
log.info("Upstream rpc detector for ${getId()} returned $rpcDetector ")
|
||||
if (rpcDetector.size != 0) {
|
||||
var changed = false
|
||||
for ((group, _) in rpcDetector) {
|
||||
if (group == "trace" || group == "debug" || group == "filter") {
|
||||
if (config.methodGroups == null) {
|
||||
config.methodGroups = UpstreamsConfig.MethodGroups(setOf("filter"), setOf())
|
||||
} else {
|
||||
val disabled = config.methodGroups!!.disabled
|
||||
val enabled = config.methodGroups!!.enabled
|
||||
if (!disabled.contains(group) && !enabled.contains(group)) {
|
||||
config.methodGroups!!.enabled = enabled.plus(group)
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
}
|
||||
rpcMethodsDetector?.detectRpcMethods()?.subscribe { rpcDetector ->
|
||||
log.info("Upstream rpc method detector for ${getId()} returned $rpcDetector ")
|
||||
if (rpcDetector.isEmpty()) {
|
||||
return@subscribe
|
||||
}
|
||||
if (changed) updateMethods(buildMethods(config, getChain()))
|
||||
if (config.methods == null) {
|
||||
config.methods = UpstreamsConfig.Methods(mutableSetOf(), mutableSetOf())
|
||||
}
|
||||
val enableMethods =
|
||||
rpcDetector
|
||||
.filter { (_, enabled) -> enabled }
|
||||
.keys
|
||||
.map { UpstreamsConfig.Method(it) }
|
||||
.toSet()
|
||||
val disableMethods =
|
||||
rpcDetector
|
||||
.filter { (_, enabled) -> !enabled }
|
||||
.keys
|
||||
.map { UpstreamsConfig.Method(it) }
|
||||
.toSet()
|
||||
config.methods =
|
||||
UpstreamsConfig.Methods(
|
||||
enableMethods
|
||||
.minus(disableMethods)
|
||||
.plus(config.methods!!.enabled),
|
||||
disableMethods
|
||||
.minus(enableMethods)
|
||||
.plus(config.methods!!.disabled),
|
||||
)
|
||||
updateMethods(buildMethods(config, getChain()))
|
||||
}
|
||||
} catch (e: RuntimeException) {
|
||||
log.error("Couldn't detect rpc modules of upstream {} due to error {}", getId(), e.message)
|
||||
} catch (e: Exception) {
|
||||
log.error("Couldn't detect methods of upstream ${getId()} due to error {}", e.message)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -285,17 +314,20 @@ open class GenericUpstream(
|
||||
validatorSubscription = validator?.start()
|
||||
?.subscribe(this::setStatus)
|
||||
}
|
||||
livenessSubscription = connector.headLivenessEvents().subscribe({
|
||||
val hasSub = it == HeadLivenessState.OK
|
||||
hasLiveSubscriptionHead.set(hasSub)
|
||||
if (it == HeadLivenessState.FATAL_ERROR) {
|
||||
headLivenessState.emitNext(UPSTREAM_FATAL_SETTINGS_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
||||
} else {
|
||||
sendUpstreamStateEvent(UPDATED)
|
||||
}
|
||||
}, {
|
||||
log.debug("Error while checking live subscription for ${getId()}", it)
|
||||
},)
|
||||
livenessSubscription = connector.headLivenessEvents().subscribe(
|
||||
{
|
||||
val hasSub = it == HeadLivenessState.OK
|
||||
hasLiveSubscriptionHead.set(hasSub)
|
||||
if (it == HeadLivenessState.FATAL_ERROR) {
|
||||
headLivenessState.emitNext(UPSTREAM_FATAL_SETTINGS_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
||||
} else {
|
||||
sendUpstreamStateEvent(UPDATED)
|
||||
}
|
||||
},
|
||||
{
|
||||
log.debug("Error while checking live subscription for ${getId()}", it)
|
||||
},
|
||||
)
|
||||
detectSettings()
|
||||
|
||||
detectLowerBlock()
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
package io.emeraldpay.dshackle.upstream.polkadot
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||
import io.emeraldpay.dshackle.upstream.ChainResponse
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetector
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.CallParams
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||
import reactor.core.publisher.Mono
|
||||
|
||||
class BasicPolkadotUpstreamRpcMethodsDetector(
|
||||
private val upstream: Upstream,
|
||||
) : UpstreamRpcMethodsDetector(upstream) {
|
||||
override fun detectByMagicMethod(): Mono<Map<String, Boolean>> =
|
||||
upstream
|
||||
.getIngressReader()
|
||||
.read(ChainRequest("rpc_methods", ListParams()))
|
||||
.flatMap(ChainResponse::requireResult)
|
||||
.map {
|
||||
Global.objectMapper
|
||||
.readValue(it, object : TypeReference<HashMap<String, List<String>>>() {})
|
||||
.getOrDefault("methods", emptyList())
|
||||
.associateWith { true }
|
||||
}.onErrorResume {
|
||||
log.warn(
|
||||
"Can't detect rpc method rpc_methods of upstream ${upstream.getId()}, reason - {}",
|
||||
it.message,
|
||||
)
|
||||
Mono.empty()
|
||||
}
|
||||
|
||||
override fun rpcMethods(): Set<Pair<String, CallParams>> = emptySet()
|
||||
}
|
||||
@@ -5,6 +5,7 @@ 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.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.data.BlockContainer
|
||||
import io.emeraldpay.dshackle.data.BlockId
|
||||
import io.emeraldpay.dshackle.foundation.ChainOptions.Options
|
||||
@@ -20,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
|
||||
import io.emeraldpay.dshackle.upstream.SingleValidator
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamRpcMethodsDetector
|
||||
import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
|
||||
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
||||
import io.emeraldpay.dshackle.upstream.calls.DefaultPolkadotMethods
|
||||
@@ -37,7 +39,6 @@ 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, api: ChainReader): Mono<BlockContainer> {
|
||||
val response = Global.objectMapper.readValue(data, PolkadotBlockResponse::class.java)
|
||||
@@ -150,6 +151,11 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() {
|
||||
override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription {
|
||||
return GenericIngressSubscription(ws, DefaultPolkadotMethods.subs.map { it.first })
|
||||
}
|
||||
|
||||
override fun upstreamRpcMethodsDetector(
|
||||
upstream: Upstream,
|
||||
config: UpstreamsConfig.Upstream<*>?,
|
||||
): UpstreamRpcMethodsDetector = BasicPolkadotUpstreamRpcMethodsDetector(upstream)
|
||||
}
|
||||
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
package io.emeraldpay.dshackle.upstream.ethereum
|
||||
|
||||
import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.reader.ChainReader
|
||||
import io.emeraldpay.dshackle.upstream.ChainCallError
|
||||
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||
import io.emeraldpay.dshackle.upstream.ChainResponse
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||
import org.assertj.core.api.Assertions
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.mockito.kotlin.doReturn
|
||||
import org.mockito.kotlin.mock
|
||||
import reactor.core.publisher.Mono
|
||||
|
||||
class BasicEthUpstreamRpcMethodsDetectorTest {
|
||||
@Test
|
||||
fun `rpc_modules without web3 and eth_getBlockReceipts`() {
|
||||
val reader =
|
||||
mock<ChainReader> {
|
||||
on {
|
||||
read(ChainRequest("rpc_modules", ListParams()))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
"""{"net": "1.0","debug": "1.0","txpool": "1.0","drpc": "1.0","erigon": "1.0","eth": "1.0","trace": "1.0"}"""
|
||||
.toByteArray(),
|
||||
null,
|
||||
),
|
||||
)
|
||||
on {
|
||||
read(ChainRequest("eth_getBlockReceipts", ListParams("latest")))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
"""[{"blockHash": "0xd12897f54acaa79f4824aa4f8e1d0f045b5568f5b942073555e9977202c5c474","blockNumber": "0x13c1108"}]"""
|
||||
.toByteArray(),
|
||||
null,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
val upstream =
|
||||
mock<Upstream> {
|
||||
on { getIngressReader() } doReturn reader
|
||||
on { getChain() } doReturn Chain.ETHEREUM__MAINNET
|
||||
}
|
||||
val config = mock<UpstreamsConfig.Upstream<*>> { }
|
||||
val detector = BasicEthUpstreamRpcMethodsDetector(upstream, config)
|
||||
Assertions.assertThat(detector.detectRpcMethods().block()).apply {
|
||||
isNotNull()
|
||||
size().isGreaterThanOrEqualTo(2)
|
||||
containsEntry("web3_clientVersion", false)
|
||||
containsValues(true)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `rpc_modules disabled and eth_getBlockReceipts`() {
|
||||
val reader =
|
||||
mock<ChainReader> {
|
||||
on {
|
||||
read(ChainRequest("rpc_modules", ListParams()))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
null,
|
||||
ChainCallError(32601, "the method rpc_modules does not exist/is not available"),
|
||||
),
|
||||
)
|
||||
on {
|
||||
read(ChainRequest("eth_getBlockReceipts", ListParams("latest")))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
"""[{"blockHash": "0xd12897f54acaa79f4824aa4f8e1d0f045b5568f5b942073555e9977202c5c474","blockNumber": "0x13c1108"}]"""
|
||||
.toByteArray(),
|
||||
null,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
val upstream =
|
||||
mock<Upstream> {
|
||||
on { getIngressReader() } doReturn reader
|
||||
on { getChain() } doReturn Chain.ETHEREUM__MAINNET
|
||||
}
|
||||
val config = mock<UpstreamsConfig.Upstream<*>> { }
|
||||
val detector = BasicEthUpstreamRpcMethodsDetector(upstream, config)
|
||||
Assertions.assertThat(detector.detectRpcMethods().block()).apply {
|
||||
isNotNull()
|
||||
hasSize(1)
|
||||
containsEntry("eth_getBlockReceipts", true)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `prefer local config methods group`() {
|
||||
val reader =
|
||||
mock<ChainReader> {
|
||||
on {
|
||||
read(ChainRequest("rpc_modules", ListParams()))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
"""{"net": "1.0","debug": "1.0","txpool": "1.0","drpc": "1.0","erigon": "1.0","eth": "1.0","trace": "1.0"}"""
|
||||
.toByteArray(),
|
||||
null,
|
||||
),
|
||||
)
|
||||
on {
|
||||
read(ChainRequest("eth_getBlockReceipts", ListParams("latest")))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
"""[{"blockHash": "0xd12897f54acaa79f4824aa4f8e1d0f045b5568f5b942073555e9977202c5c474","blockNumber": "0x13c1108"}]"""
|
||||
.toByteArray(),
|
||||
null,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
val upstream =
|
||||
mock<Upstream> {
|
||||
on { getIngressReader() } doReturn reader
|
||||
on { getChain() } doReturn Chain.ETHEREUM__MAINNET
|
||||
}
|
||||
val config =
|
||||
mock<UpstreamsConfig.Upstream<*>> {
|
||||
on { methodGroups } doReturn
|
||||
UpstreamsConfig.MethodGroups(
|
||||
emptySet(),
|
||||
setOf("eth"),
|
||||
)
|
||||
}
|
||||
val detector = BasicEthUpstreamRpcMethodsDetector(upstream, config)
|
||||
Assertions.assertThat(detector.detectRpcMethods().block()).apply {
|
||||
isNotNull()
|
||||
containsEntry("eth_getBlockByNumber", false)
|
||||
containsEntry("eth_getBlockReceipts", true)
|
||||
containsEntry("debug_traceBlock", true)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package io.emeraldpay.dshackle.upstream.polkadot
|
||||
|
||||
import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.reader.ChainReader
|
||||
import io.emeraldpay.dshackle.upstream.ChainCallError
|
||||
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||
import io.emeraldpay.dshackle.upstream.ChainResponse
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||
import org.assertj.core.api.Assertions
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.mockito.kotlin.doReturn
|
||||
import org.mockito.kotlin.mock
|
||||
import reactor.core.publisher.Mono
|
||||
|
||||
class BasicPolkadotUpstreamRpcMethodsDetectorTest {
|
||||
@Test
|
||||
fun `rpc_methods enabled`() {
|
||||
val reader =
|
||||
mock<ChainReader> {
|
||||
on {
|
||||
read(ChainRequest("rpc_methods", ListParams()))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
""" {"methods": ["account_nextIndex","archive_unstable_body","archive_unstable_call"]}"""
|
||||
.toByteArray(),
|
||||
null,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
val upstream =
|
||||
mock<Upstream> {
|
||||
on { getIngressReader() } doReturn reader
|
||||
on { getChain() } doReturn Chain.POLKADOT__MAINNET
|
||||
}
|
||||
val detector = BasicPolkadotUpstreamRpcMethodsDetector(upstream)
|
||||
Assertions.assertThat(detector.detectRpcMethods().block()).apply {
|
||||
isNotNull()
|
||||
hasSize(3)
|
||||
containsKeys("account_nextIndex", "archive_unstable_body", "archive_unstable_call")
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `rpc_methods disabled`() {
|
||||
val reader =
|
||||
mock<ChainReader> {
|
||||
on {
|
||||
read(ChainRequest("rpc_methods", ListParams()))
|
||||
} doReturn
|
||||
Mono.just(
|
||||
ChainResponse(
|
||||
null,
|
||||
ChainCallError(32601, "the method rpc_methods does not exist/is not available"),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
val upstream =
|
||||
mock<Upstream> {
|
||||
on { getIngressReader() } doReturn reader
|
||||
on { getChain() } doReturn Chain.POLKADOT__MAINNET
|
||||
}
|
||||
val detector = BasicPolkadotUpstreamRpcMethodsDetector(upstream)
|
||||
Assertions.assertThat(detector.detectRpcMethods().block()).isNull()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user