diff --git a/build.gradle b/build.gradle index d4f15101..76d750de 100644 --- a/build.gradle +++ b/build.gradle @@ -123,8 +123,16 @@ dependencies { implementation(variantOf(libs.netty.tcnative.boringssl) { classifier("linux-x86_64") }) implementation(variantOf(libs.netty.tcnative.boringssl) { classifier("osx-x86_64") }) implementation 'dshackle:foundation:1.0.0' + + implementation files('src/main/resources/LogsOracle.jar') } +// Enable 'Foreign Function & Memory API' (JEP 434) +// Drop after update on Java 21 +tasks.withType(JavaCompile) { options.compilerArgs += "--enable-preview" } +tasks.withType(Test) { jvmArgs += "--enable-preview" } +tasks.withType(JavaExec) { jvmArgs += "--enable-preview" } + compileKotlin { compilerOptions.jvmTarget.set(JvmTarget.JVM_20) } @@ -327,4 +335,4 @@ ktlint { } } -compileKotlin.dependsOn chainscodegen \ No newline at end of file +compileKotlin.dependsOn chainscodegen diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt index b33ab61b..da05fbca 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.config.CacheConfig import io.emeraldpay.dshackle.config.ChainsConfig import io.emeraldpay.dshackle.config.CompressionConfig import io.emeraldpay.dshackle.config.HealthConfig +import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.MainConfig import io.emeraldpay.dshackle.config.MainConfigReader import io.emeraldpay.dshackle.config.MonitoringConfig @@ -130,6 +131,11 @@ open class Config( return mainConfig.cache ?: CacheConfig() } + @Bean + open fun indexConfig(@Autowired mainConfig: MainConfig): IndexConfig { + return mainConfig.index ?: IndexConfig() + } + @Bean open fun signatureConfig(@Autowired mainConfig: MainConfig): SignatureConfig { return mainConfig.signature ?: SignatureConfig() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfig.kt new file mode 100644 index 00000000..00202c6e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfig.kt @@ -0,0 +1,21 @@ +package io.emeraldpay.dshackle.config + +import io.emeraldpay.dshackle.Chain + +class IndexConfig { + var items: HashMap = HashMap() + + class Index( + var rpc: String, + var store: String, + var ram_limit: Long?, + ) + + fun isChainEnabled(chain: Chain): Boolean { + return items.containsKey(chain) + } + + fun getByChain(chain: Chain): Index? { + return items.get(chain) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfigReader.kt new file mode 100644 index 00000000..1297670d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfigReader.kt @@ -0,0 +1,62 @@ +package io.emeraldpay.dshackle.config + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.foundation.YamlConfigReader +import org.apache.commons.lang3.StringUtils +import org.slf4j.LoggerFactory +import org.springframework.util.unit.DataSize +import org.yaml.snakeyaml.nodes.MappingNode + +class IndexConfigReader : YamlConfigReader() { + + companion object { + private val log = LoggerFactory.getLogger(IndexConfigReader::class.java) + } + + override fun read(input: MappingNode?): IndexConfig? { + return getList(input, "index")?.let { items -> + val config = IndexConfig() + + items.value.map { + val blockchainRaw = getValueAsString(it, "chain") + if (blockchainRaw == null || StringUtils.isEmpty(blockchainRaw) || Global.chainById(blockchainRaw) == Chain.UNSPECIFIED) { + throw InvalidConfigYamlException(filename, it.startMark, "Invalid blockchain or not specified") + } + + val blockchain = Global.chainById(blockchainRaw) + if (config.items.containsKey(blockchain)) { + throw InvalidConfigYamlException(filename, it.startMark, "Duplicated indexes") + } + + val rpc = getValueAsString(it, "rpc") + if (rpc == null || StringUtils.isEmpty(rpc)) { + throw InvalidConfigYamlException(filename, it.startMark, "Invalid rpc specified") + } + + val store = getValueAsString(it, "store") + if (store == null || StringUtils.isEmpty(store)) { + throw InvalidConfigYamlException(filename, it.startMark, "Invalid store directory or not specified") + } + + val limit = getMapping(it, "limit") + val ram_limit = limit?.let { + val raw = getValueAsString(limit, "ram") + if (raw == null || StringUtils.isEmpty(raw)) { + return null + } + + try { + DataSize.parse(raw).toBytes() + } catch (e: IllegalArgumentException) { + throw InvalidConfigYamlException(filename, it.startMark, "Invalid limit for index") + } + } + + config.items.put(blockchain, IndexConfig.Index(rpc, store, ram_limit)) + } + + config + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt index 9d6b125a..9146b376 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfig.kt @@ -21,6 +21,7 @@ class MainConfig { var tls: AuthConfig.ServerTlsAuth? = null var passthrough: Boolean = false var cache: CacheConfig? = null + var index: IndexConfig? = null var proxy: ProxyConfig? = null var upstreams: UpstreamsConfig? = null var tokens: TokensConfig? = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt index 90cdef65..7522c562 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt @@ -29,6 +29,7 @@ class MainConfigReader( private val optionsReader = ChainOptionsReader() private val upstreamsConfigReader = UpstreamsConfigReader(fileResolver, optionsReader) private val cacheConfigReader = CacheConfigReader() + private val indexConfigReader = IndexConfigReader() private val tokensConfigReader = TokensConfigReader() private val monitoringConfigReader = MonitoringConfigReader() private val accessLogReader = AccessLogReader() @@ -63,6 +64,9 @@ class MainConfigReader( cacheConfigReader.read(input)?.let { config.cache = it } + indexConfigReader.read(input)?.let { + config.index = it + } tokensConfigReader.read(input)?.let { config.tokens = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt index daa38e6a..dc0a299d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.config.context import io.emeraldpay.dshackle.BlockchainType.BITCOIN import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.cache.CachesFactory +import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.upstream.CallTargetsHolder import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream @@ -26,6 +27,9 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) headScheduler: Scheduler, tracer: Tracer, multistreamEventsScheduler: Scheduler, + indexConfig: IndexConfig, + @Qualifier("logsOracleScheduler") + logsOracleScheduler: Scheduler, ): List { return Chain.entries .filterNot { it == Chain.UNSPECIFIED } @@ -33,7 +37,15 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) if (chain.type == BITCOIN) { bitcoinMultistream(chain, cachesFactory, headScheduler, multistreamEventsScheduler) } else { - genericMultistream(chain, cachesFactory, headScheduler, tracer, multistreamEventsScheduler) + genericMultistream( + chain, + cachesFactory, + headScheduler, + tracer, + multistreamEventsScheduler, + indexConfig.getByChain(chain), + logsOracleScheduler, + ) } } } @@ -44,6 +56,8 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) headScheduler: Scheduler, tracer: Tracer, multistreamEventsScheduler: Scheduler, + logsOracleConfig: IndexConfig.Index?, + logsOracleScheduler: Scheduler, ): Multistream { val name = "multi-$chain" val cs = ChainSpecificRegistry.resolve(chain) @@ -58,6 +72,8 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) cs.makeCachingReaderBuilder(tracer), cs::localReaderBuilder, cs.subscriptionBuilder(headScheduler), + logsOracleConfig, + logsOracleScheduler, ).also { register(it, name) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt index ac2892e9..a5383564 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt @@ -30,6 +30,11 @@ open class SchedulersConfig { return makeScheduler("head-scheduler", 4, monitoringConfig) } + @Bean + open fun logsOracleScheduler(monitoringConfig: MonitoringConfig): Scheduler { + return makeScheduler("logs-oracle", 4, monitoringConfig) + } + @Bean open fun multistreamEventsScheduler(monitoringConfig: MonitoringConfig): Scheduler { return makeScheduler("events-scheduler", 4, monitoringConfig) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt index c6c5ccc3..4836420e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.startup.configure import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.startup.QuorumForLabels @@ -24,11 +25,12 @@ import java.util.concurrent.atomic.AtomicInteger @Component class BitcoinUpstreamCreator( chainsConfig: ChainsConfig, + indexConfig: IndexConfig, callTargets: CallTargetsHolder, private val genericConnectorFactoryCreator: ConnectorFactoryCreator, private val fileResolver: FileResolver, private val headScheduler: Scheduler, -) : UpstreamCreator(chainsConfig, callTargets) { +) : UpstreamCreator(chainsConfig, indexConfig, callTargets) { private var seq = AtomicInteger(0) override fun createUpstream( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt index 3f653db8..211a9219 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.startup.configure import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.upstream.CallTargetsHolder @@ -11,9 +12,10 @@ import org.springframework.stereotype.Component @Component class EthereumUpstreamCreator( chainsConfig: ChainsConfig, + indexConfig: IndexConfig, callTargets: CallTargetsHolder, genericConnectorFactoryCreator: ConnectorFactoryCreator, -) : GenericUpstreamCreator(chainsConfig, callTargets, genericConnectorFactoryCreator) { +) : GenericUpstreamCreator(chainsConfig, indexConfig, callTargets, genericConnectorFactoryCreator) { override fun createUpstream( upstreamsConfig: UpstreamsConfig.Upstream<*>, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt index 9d13939a..eff68248 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.startup.configure import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.startup.QuorumForLabels @@ -19,9 +20,10 @@ import kotlin.math.abs @Component open class GenericUpstreamCreator( chainsConfig: ChainsConfig, + indexConfig: IndexConfig, callTargets: CallTargetsHolder, private val genericConnectorFactoryCreator: ConnectorFactoryCreator, -) : UpstreamCreator(chainsConfig, callTargets) { +) : UpstreamCreator(chainsConfig, indexConfig, callTargets) { private val hashes: MutableMap = HashMap() override fun createUpstream( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt index 8eba6746..786b9a88 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.startup.configure import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.upstream.CallTargetsHolder @@ -14,6 +15,7 @@ import org.slf4j.LoggerFactory abstract class UpstreamCreator( private val chainsConfig: ChainsConfig, + private val indexConfig: IndexConfig, private val callTargets: CallTargetsHolder, ) { protected val log: Logger = LoggerFactory.getLogger(this::class.java) @@ -45,7 +47,7 @@ abstract class UpstreamCreator( protected fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods { return if (config.methods != null || config.methodGroups != null) { ManagedCallMethods( - delegate = callTargets.getDefaultMethods(chain), + delegate = callTargets.getDefaultMethods(chain, indexConfig.isChainEnabled(chain)), enabled = config.methods?.enabled?.map { it.name }?.toSet() ?: emptySet(), disabled = config.methods?.disabled?.map { it.name }?.toSet() ?: emptySet(), groupsEnabled = config.methodGroups?.enabled ?: emptySet(), @@ -61,7 +63,7 @@ abstract class UpstreamCreator( } } } else { - callTargets.getDefaultMethods(chain) + callTargets.getDefaultMethods(chain, indexConfig.isChainEnabled(chain)) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt index 3f2e51af..56bddb33 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt @@ -18,14 +18,14 @@ import org.springframework.stereotype.Component class CallTargetsHolder { private val callTargets = HashMap() - fun getDefaultMethods(chain: Chain): CallMethods { - return callTargets[chain] ?: return setupDefaultMethods(chain) + fun getDefaultMethods(chain: Chain, hasLogsOracle: Boolean): CallMethods { + return callTargets[chain] ?: return setupDefaultMethods(chain, hasLogsOracle) } - private fun setupDefaultMethods(chain: Chain): CallMethods { + private fun setupDefaultMethods(chain: Chain, hasLogsOracle: Boolean): CallMethods { val created = when (chain.type) { BITCOIN -> DefaultBitcoinMethods() - ETHEREUM -> DefaultEthereumMethods(chain) + ETHEREUM -> DefaultEthereumMethods(chain, hasLogsOracle) STARKNET -> DefaultStarknetMethods(chain) POLKADOT -> DefaultPolkadotMethods() SOLANA -> DefaultSolanaMethods() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt new file mode 100644 index 00000000..5a14f9d3 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt @@ -0,0 +1,42 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.config.IndexConfig +import org.slf4j.LoggerFactory +import reactor.core.Disposable +import reactor.core.publisher.Mono +import reactor.core.scheduler.Scheduler + +class LogsOracle( + private val config: IndexConfig.Index, + private val upstream: Multistream, + private val scheduler: Scheduler, +) { + + private val log = LoggerFactory.getLogger(LogsOracle::class.java) + + private var subscription: Disposable? = null + private val db = org.drpc.logsoracle.LogsOracle(config.store, config.store, config.ram_limit ?: 0L) + + fun start() { + subscription = upstream.getHead().getFlux() + .doOnError { t -> log.warn("Failed to subscribe head for oracle", t) } + .subscribe { println(it.height); db.UpdateHeight(it.height) } + } + + fun stop() { + db.close() + + subscription?.dispose() + subscription = null + } + + fun estimate( + fromBlock: Long?, + toBlock: Long?, + address: List, + topics: List>, + ): Mono { + return Mono.fromCallable { db.Query(fromBlock, toBlock, address, topics) } + .subscribeOn(scheduler) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt index 2f52c9c3..885b90ea 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt @@ -32,6 +32,7 @@ import io.emeraldpay.etherjar.rpc.RpcException */ class DefaultEthereumMethods( private val chain: Chain, + private val hasLogsOracle: Boolean = false, ) : CallMethods { private val version = "\"EmeraldDshackle/${Global.version}\"" @@ -137,6 +138,7 @@ class DefaultEthereumMethods( specialMethods + headVerifiedMethods - chainUnsupportedMethods(chain) + + getDrpcVendorMethods(chain) + getChainSpecificMethods(chain) } @@ -148,6 +150,7 @@ class DefaultEthereumMethods( firstValueMethods.contains(method) -> AlwaysQuorum() anyResponseMethods.contains(method) -> NotLaggingQuorum(4) headVerifiedMethods.contains(method) -> NotLaggingQuorum(1) + getDrpcVendorMethods(chain).contains(method) -> AlwaysQuorum() possibleNotIndexedMethods.contains(method) -> NotNullQuorum() specialMethods.contains(method) -> { when (method) { @@ -174,6 +177,15 @@ class DefaultEthereumMethods( } } + private fun getDrpcVendorMethods(chain: Chain): List { + val supported = mutableListOf() + + // Currently tested on eth mainnet only, should potentially work for all compatible ones. + if (chain == Chain.ETHEREUM__MAINNET && hasLogsOracle) { supported.add("drpc_getLogsEstimate") } + + return supported + } + private fun getChainSpecificMethods(chain: Chain): List { return when (chain) { Chain.OPTIMISM__MAINNET, Chain.OPTIMISM__GOERLI -> listOf( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt index 648d9dd7..58201b99 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt @@ -11,6 +11,7 @@ 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.Upstream import io.emeraldpay.dshackle.upstream.UpstreamValidator @@ -48,8 +49,9 @@ object EthereumChainSpecific : AbstractPollChainSpecific() { cachingReader: CachingReader, methods: CallMethods, head: Head, + logsOracle: LogsOracle?, ): Mono { - return Mono.just(EthereumLocalReader(cachingReader as EthereumCachingReader, methods, head)) + return Mono.just(EthereumLocalReader(cachingReader as EthereumCachingReader, methods, head, logsOracle)) } override fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt index 26212358..24c4a2cb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.JsonRpcReader import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.LogsOracle import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -41,6 +42,7 @@ class EthereumLocalReader( private val reader: EthereumCachingReader, private val methods: CallMethods, private val head: Head, + private val logsOracle: LogsOracle?, ) : JsonRpcReader { override fun read(key: JsonRpcRequest): Mono { @@ -120,6 +122,9 @@ class EthereumLocalReader( .read(hash) .map { it.data to it.upstreamId } } + method == "drpc_getLogsEstimate" -> { + getLogsEstimate(params) + } else -> null } } @@ -167,4 +172,82 @@ class EthereumLocalReader( return reader.blocksByHeightAsCont() .read(number).map { it.data.json!! to it.upstreamId } } + + fun getLogsEstimate(params: List): Mono>? { + if (logsOracle == null) { + throw NotImplementedError() + } + if (params.size != 1 || params[0] == null) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameters") + } + + val req = params[0] as LinkedHashMap + + val fromBlock = try { parseBlockRef(req.get("fromBlock") as String?) } catch (_: IllegalArgumentException) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'fromBlock' parameter") + } + val toBlock = try { parseBlockRef(req.get("toBlock") as String?) } catch (_: IllegalArgumentException) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'toBlock' parameter") + } + val address: List = try { + val it = req.get("address") ?: listOf() + + if (it is String) { + listOf(it) + } else { + it as List + } + } catch (_: Exception) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'address' parameter") + } + val topics: List> = try { + val tpcs = req.get("topics")?.let { it as List } ?: listOf() + if (tpcs.size > 4) { + throw IllegalArgumentException() + } + + tpcs.map { + if (it is String) { + listOf(it) + } else { + it as List + } + } + } catch (_: Exception) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'topics' parameter") + } + + return logsOracle.estimate(fromBlock, toBlock, address, topics) + .map { it.toString().toByteArray() to null } + } + + private fun parseBlockRef(blockRef: String?): Long? { + when { + blockRef == null -> { + return null + } + blockRef == "latest" -> { + return head.getCurrentHeight() ?: return null + } + blockRef == "earliest" -> { + return 0 + } + blockRef == "finalized" || blockRef == "safe" || blockRef == "pending" -> { + return null + } + blockRef.startsWith("0x") -> { + val quantity = HexQuantity.from(blockRef) ?: throw IllegalArgumentException() + return quantity.value.let { + if (it < BigInteger.valueOf(Long.MAX_VALUE) && it >= BigInteger.ZERO) { + it.toLong() + } else { + throw IllegalArgumentException() + } + } + } + else -> { + throw IllegalArgumentException() + } + } + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt index 840e2477..fba52c94 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt @@ -12,6 +12,7 @@ import io.emeraldpay.dshackle.upstream.EmptyEgressSubscription 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.NoIngressSubscription import io.emeraldpay.dshackle.upstream.NoopCachingReader @@ -31,6 +32,7 @@ abstract class AbstractChainSpecific : ChainSpecific { cachingReader: CachingReader, methods: CallMethods, head: Head, + logsOracle: LogsOracle?, ): Mono { return Mono.just(LocalReader(methods)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt index c613c88a..9249f660 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -17,6 +17,7 @@ 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.Upstream import io.emeraldpay.dshackle.upstream.UpstreamValidator @@ -34,7 +35,7 @@ import reactor.core.publisher.Mono import reactor.core.scheduler.Scheduler typealias SubscriptionBuilder = (Multistream) -> EgressSubscription -typealias LocalReaderBuilder = (CachingReader, CallMethods, Head) -> Mono +typealias LocalReaderBuilder = (CachingReader, CallMethods, Head, LogsOracle?) -> Mono typealias CachingReaderBuilder = (Multistream, Caches, Factory) -> CachingReader interface ChainSpecific { @@ -46,7 +47,7 @@ interface ChainSpecific { fun unsubscribeNewHeadsRequest(subId: String): JsonRpcRequest - fun localReaderBuilder(cachingReader: CachingReader, methods: CallMethods, head: Head): Mono + fun localReaderBuilder(cachingReader: CachingReader, methods: CallMethods, head: Head, logsOracle: LogsOracle?): Mono fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt index 5e9b03e8..c83cf6bc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt @@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.upstream.generic import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.JsonRpcReader @@ -31,6 +32,7 @@ import io.emeraldpay.dshackle.upstream.EmptyHead import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.HeadLagObserver import io.emeraldpay.dshackle.upstream.Lifecycle +import io.emeraldpay.dshackle.upstream.LogsOracle import io.emeraldpay.dshackle.upstream.MergedHead import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector @@ -58,6 +60,8 @@ open class GenericMultistream( cachingReaderBuilder: CachingReaderBuilder, private val localReaderBuilder: LocalReaderBuilder, private val subscriptionBuilder: SubscriptionBuilder, + logsOracleConfig: IndexConfig.Index? = null, + private val logsOracleScheduler: Scheduler, ) : Multistream(chain, caches, callSelector, multistreamEventsScheduler) { private val cachingReader = cachingReaderBuilder(this, caches, getMethodsFactory()) @@ -76,6 +80,10 @@ open class GenericMultistream( headScheduler, ) + private val logsOracle: LogsOracle? = logsOracleConfig?.let { + LogsOracle(logsOracleConfig, this, logsOracleScheduler) + } + private var subscription: EgressSubscription = subscriptionBuilder(this) private val filteredHeads: MutableMap = @@ -86,12 +94,14 @@ open class GenericMultistream( head.start() onHeadUpdated(head) cachingReader.start() + logsOracle?.start() } override fun stop() { super.stop() cachingReader.stop() filteredHeads.clear() + logsOracle?.stop() } override fun addHead(upstream: Upstream) { @@ -181,7 +191,7 @@ open class GenericMultistream( } override fun getLocalReader(): Mono { - return localReaderBuilder(cachingReader, getMethods(), getHead()) + return localReaderBuilder(cachingReader, getMethods(), getHead(), logsOracle) } override fun getEgressSubscription(): EgressSubscription { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt index 48d05ae9..1bdb96d5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt @@ -14,6 +14,7 @@ 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.Upstream @@ -74,6 +75,7 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() { cachingReader: CachingReader, methods: CallMethods, head: Head, + logsOracle: LogsOracle?, ): Mono { return Mono.just(LocalReader(methods)) } diff --git a/src/main/resources/LogsOracle.jar b/src/main/resources/LogsOracle.jar new file mode 100644 index 00000000..7194f50d Binary files /dev/null and b/src/main/resources/LogsOracle.jar differ diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 06fb63bb..04e36064 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -398,7 +398,7 @@ class NativeCallSpec extends Specification { def "Prepare call adds height selector for not-lagging quorum"() { setup: def methods = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), ["foo_bar"] as Set, [] as Set, [] as Set, [] as Set ) methods.setQuorum("foo_bar", "not_lagging") @@ -439,7 +439,7 @@ class NativeCallSpec extends Specification { def "Prepare call adds decorator for eth_newFilter"() { setup: def methods = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), ["eth_newFilter"] as Set, [] as Set, [] as Set, [] as Set ) methods.setQuorum("eth_newFilter", "always") @@ -470,7 +470,7 @@ class NativeCallSpec extends Specification { def "Prepare call adds decorator for eth_getFilterChanges"() { setup: def methods = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), ["eth_getFilterChanges"] as Set, [] as Set, [] as Set, [] as Set ) def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, TestingCommons.upstream()) @@ -500,7 +500,7 @@ class NativeCallSpec extends Specification { def "Prepare call adds decorator for eth_uninstallFilter"() { setup: def methods = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), ["eth_uninstallFilter"] as Set, [] as Set, [] as Set, [] as Set ) def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, TestingCommons.upstream()) @@ -600,7 +600,7 @@ class NativeCallSpec extends Specification { } def quorum = new AlwaysQuorum() def methods = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), [] as Set, [] as Set, ["filter"] as Set, [] as Set ) def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, new ArrayList()) @@ -636,7 +636,7 @@ class NativeCallSpec extends Specification { } def quorum = new AlwaysQuorum() def methods = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), [] as Set, [] as Set, ["filter"] as Set, [] as Set ) def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, new ArrayList()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy index 56da1932..1ce79a88 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy @@ -36,7 +36,7 @@ class GenericUpstreamMock extends GenericUpstream { static CallMethods allMethods() { new AggregatedCallMethods([ - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), new DefaultBitcoinMethods(), new DirectCallMethods(["eth_test"]) ]) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 5cf1e836..d88a8403 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -51,7 +51,8 @@ class MultistreamHolderMock implements MultistreamHolder { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()) + io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), + null, Schedulers.immediate() ) upstreams[chain].addUpstream(up) } else { @@ -104,7 +105,7 @@ class MultistreamHolderMock implements MultistreamHolder { super(chain, Schedulers.immediate(), null, upstreams, caches, Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(new BraveTracer(null, null, null)), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) } EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 3afa1302..288cddf2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -98,6 +98,8 @@ class TestingCommons { EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), + null, + Schedulers.immediate() ).tap { it.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED) @@ -124,14 +126,18 @@ class TestingCommons { return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), + null, + Schedulers.immediate()) } static Multistream multistreamClassicWithoutUpstreams(Chain chain) { return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), + null, + Schedulers.immediate()) } static FileResolver fileResolver() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 3a629d4e..056993a7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -39,7 +39,7 @@ import static java.util.List.of class FilteredApisSpec extends Specification { - def ethereumTargets = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def ethereumTargets = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) def "Verifies labels"() { setup: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index c57129eb..594b00c7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -59,7 +59,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() @@ -194,7 +194,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) expect: multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock) @@ -267,7 +267,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) when: ms.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) @@ -298,7 +298,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) def head1 = createBlock(250, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448401b") def head2 = createBlock(270, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448402b") def head3 = createBlock(100, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448412b") @@ -333,7 +333,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) multistream.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) ) @@ -371,7 +371,7 @@ class MultistreamSpec extends Specification { Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, - StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) + StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), null, Schedulers.immediate()) } @NotNull diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethodsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethodsSpec.groovy index f952c5d3..ddf3b128 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethodsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethodsSpec.groovy @@ -7,7 +7,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "eth_chainId is available"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) when: def act = methods.isAvailable("eth_chainId") then: @@ -16,7 +16,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "eth_chainId is hardcoded"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) when: def act = methods.isHardcoded("eth_chainId") then: @@ -25,7 +25,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "eth_chainId is not callable"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) when: def act = methods.isCallable("eth_chainId") then: @@ -34,7 +34,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "Provides hardcoded correct chainId"() { expect: - new String(new DefaultEthereumMethods(chain).executeHardcoded("eth_chainId")) == id + new String(new DefaultEthereumMethods(chain, false).executeHardcoded("eth_chainId")) == id where: chain | id Chain.ETHEREUM__MAINNET | '"0x1"' @@ -44,7 +44,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "Optimism chain unsupported methods"() { setup: - def methods = new DefaultEthereumMethods(Chain.OPTIMISM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.OPTIMISM__MAINNET, false) when: def acc = methods.isAvailable("eth_getAccounts") def trans = methods.isAvailable("eth_sendTransaction") @@ -55,7 +55,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "Has supported specific methods"() { expect: - new DefaultEthereumMethods(chain).getSupportedMethods().containsAll(methods) + new DefaultEthereumMethods(chain, false).getSupportedMethods().containsAll(methods) where: chain | methods Chain.POLYGON__MAINNET | ["bor_getAuthor", @@ -69,7 +69,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "Has no filter methods by default"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) when: def act = methods.getSupportedMethods().findAll { it.containsIgnoreCase("filter") } then: @@ -78,7 +78,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "Has no trace methods by default"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) when: def act = methods.getSupportedMethods().findAll { it.containsIgnoreCase("trace") } then: @@ -87,7 +87,7 @@ class DefaultEthereumMethodsSpec extends Specification { def "Default eth methods are available"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) expect: methods.isAvailable(method) where: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy index fa4c1de4..a5350d19 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy @@ -94,7 +94,7 @@ class ManagedCallMethodsSpec extends Specification { def "Use custom quorum if provided"() { setup: def managed = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), ["eth_test", "eth_foo", "eth_bar"] as Set, [] as Set, [] as Set, @@ -120,7 +120,7 @@ class ManagedCallMethodsSpec extends Specification { def "Doesn't reuse same instance"() { def managed = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), ["eth_test"] as Set, [] as Set, [] as Set, @@ -145,7 +145,7 @@ class ManagedCallMethodsSpec extends Specification { def "Test enable method group"() { setup: def managed = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), [] as Set, [] as Set, ["filter"] as Set, @@ -169,7 +169,7 @@ class ManagedCallMethodsSpec extends Specification { def "Test enable method group minus one"() { setup: def managed = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), [] as Set, ["eth_newPendingTransactionFilter"] as Set, ["filter"] as Set, @@ -193,7 +193,7 @@ class ManagedCallMethodsSpec extends Specification { def "Test disabled group not disable enabled method"() { setup: def managed = new ManagedCallMethods( - new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), + new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), ["eth_newPendingTransactionFilter"] as Set, [] as Set, [] as Set, diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index ff4d1704..43d541ce 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -43,7 +43,7 @@ class EthereumDirectReaderSpec extends Specification { transactions = [] } def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -70,7 +70,7 @@ class EthereumDirectReaderSpec extends Specification { def "Produce empty result on non-existing block"() { setup: def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -103,7 +103,7 @@ class EthereumDirectReaderSpec extends Specification { transactions = [] } def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -136,7 +136,7 @@ class EthereumDirectReaderSpec extends Specification { blockHash = BlockHash.from(hash1) } def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -169,7 +169,7 @@ class EthereumDirectReaderSpec extends Specification { blockHash = BlockHash.from(hash1) } def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -199,7 +199,7 @@ class EthereumDirectReaderSpec extends Specification { blockHash = BlockHash.from(hash1) } def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } def caches = Mock(Caches) { // note that the Caches needs a Height value, otherwise it's not cached @@ -228,7 +228,7 @@ class EthereumDirectReaderSpec extends Specification { def "Produce empty on non-existing tx"() { setup: def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -258,7 +258,7 @@ class EthereumDirectReaderSpec extends Specification { } } def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -289,7 +289,7 @@ class EthereumDirectReaderSpec extends Specification { } } def calls = Mock(Factory) { - 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -323,7 +323,7 @@ class EthereumDirectReaderSpec extends Specification { transactions = [] } def calls = Mock(Factory) { - 3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } def result = Mono.just( new RpcReader.Result( @@ -363,7 +363,7 @@ class EthereumDirectReaderSpec extends Specification { transactions = [] } def calls = Mock(Factory) { - 3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } def result = Mono.just( new RpcReader.Result( @@ -400,7 +400,7 @@ class EthereumDirectReaderSpec extends Specification { } } def calls = Mock(Factory) { - 4 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + 4 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) } EthereumDirectReader reader = new EthereumDirectReader( up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy index 0807ce58..f4fd4602 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy @@ -21,16 +21,17 @@ class EthereumLocalReaderSpec extends Specification { def "Calls hardcoded"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) def router = new EthereumLocalReader( new EthereumCachingReader( TestingCommons.multistream(TestingCommons.api()), Caches.default(), - ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)), + ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false)), TestingCommons.tracerMock() ), methods, - new EmptyHead() + new EmptyHead(), + null ) when: def act = router.read(new JsonRpcRequest("eth_coinbase", [])).block(Duration.ofSeconds(1)) @@ -40,16 +41,17 @@ class EthereumLocalReaderSpec extends Specification { def "Returns empty if nonce set"() { setup: - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) def router = new EthereumLocalReader( new EthereumCachingReader( TestingCommons.multistream(TestingCommons.api()), Caches.default(), - ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)), + ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false)), TestingCommons.tracerMock() ), methods, - new EmptyHead() + new EmptyHead(), + null ) when: def act = router.read(new JsonRpcRequest("eth_getTransactionByHash", ["test"], 10)) @@ -72,8 +74,8 @@ class EthereumLocalReaderSpec extends Specification { ) } } - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) - def router = new EthereumLocalReader(reader, methods, head) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) + def router = new EthereumLocalReader(reader, methods, head, null) when: def act = router.getBlockByNumber(["latest", false]) @@ -100,8 +102,8 @@ class EthereumLocalReaderSpec extends Specification { ) } } - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) - def router = new EthereumLocalReader(reader, methods, head) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) + def router = new EthereumLocalReader(reader, methods, head, null) when: def act = router.getBlockByNumber(["earliest", false]) @@ -128,8 +130,8 @@ class EthereumLocalReaderSpec extends Specification { ) } } - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) - def router = new EthereumLocalReader(reader, methods, head) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) + def router = new EthereumLocalReader(reader, methods, head, null) when: def act = router.getBlockByNumber(["0x123ef", false]) @@ -152,8 +154,8 @@ class EthereumLocalReaderSpec extends Specification { _ * txByHashAsCont() >> new EmptyReader<>() _ * blocksByHeightAsCont() >> new EmptyReader<>() } - def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) - def router = new EthereumLocalReader(reader, methods, head) + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) + def router = new EthereumLocalReader(reader, methods, head, null) when: def act = router.getBlockByNumber(["0x0", true]) diff --git a/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt index 4750c1ab..13af79d6 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt @@ -179,6 +179,8 @@ class ReloadConfigTest { cs.makeCachingReaderBuilder(mock()), cs::localReaderBuilder, cs.subscriptionBuilder(Schedulers.boundedElastic()), + null, + Schedulers.fromExecutor(Executors.newFixedThreadPool(6)), ) }