Integrate drpc-logs-oracle (#1096) (#335)

* add oracle jar to build (#1096) * add oracle config (#1096) * add eth_getLogsEstimate (#1096) * add scheduler for oracle (#1096) * fix tests :[ (#1096)
2023-11-22 13:45:00 +01:00
parent a7f74bb71a
commit 5a154ac817
33 changed files with 369 additions and 73 deletions
--- 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()
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfig.kt
+++ 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<Chain, Index> = HashMap<Chain, Index>()
+
+    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)
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/IndexConfigReader.kt
+++ 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<IndexConfig>() {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(IndexConfigReader::class.java)
+    }
+
+    override fun read(input: MappingNode?): IndexConfig? {
+        return getList<MappingNode>(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
+        }
+    }
+}
--- 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
--- 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
        }
--- 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<Multistream> {
        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) }
    }

--- 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)
--- 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(
--- 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<*>,
--- 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<Byte, Boolean> = HashMap()

    override fun createUpstream(
--- 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))
        }
    }
 }
--- 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<Chain, CallMethods>()

-    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()
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LogsOracle.kt
+++ 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<String>,
+        topics: List<List<String>>,
+    ): Mono<Long> {
+        return Mono.fromCallable { db.Query(fromBlock, toBlock, address, topics) }
+            .subscribeOn(scheduler)
+    }
+}
--- 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<String> {
+        val supported = mutableListOf<String>()
+
+        // 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<String> {
        return when (chain) {
            Chain.OPTIMISM__MAINNET, Chain.OPTIMISM__GOERLI -> listOf(
--- 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<JsonRpcReader> {
-        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 {
--- 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<JsonRpcResponse> {
@@ -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<Any?>): Mono<Pair<ByteArray, String?>>? {
+        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<String, Any?>
+
+        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<String> = try {
+            val it = req.get("address") ?: listOf<String>()
+
+            if (it is String) {
+                listOf<String>(it)
+            } else {
+                it as List<String>
+            }
+        } catch (_: Exception) {
+            throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Invalid 'address' parameter")
+        }
+        val topics: List<List<String>> = try {
+            val tpcs = req.get("topics")?.let { it as List<Any> } ?: listOf<Any>()
+            if (tpcs.size > 4) {
+                throw IllegalArgumentException()
+            }
+
+            tpcs.map {
+                if (it is String) {
+                    listOf<String>(it)
+                } else {
+                    it as List<String>
+                }
+            }
+        } 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()
+            }
+        }
+    }
 }
--- 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<JsonRpcReader> {
        return Mono.just(LocalReader(methods))
    }
--- 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<JsonRpcReader>
+typealias LocalReaderBuilder = (CachingReader, CallMethods, Head, LogsOracle?) -> Mono<JsonRpcReader>
 typealias CachingReaderBuilder = (Multistream, Caches, Factory<CallMethods>) -> CachingReader

 interface ChainSpecific {
@@ -46,7 +47,7 @@ interface ChainSpecific {

    fun unsubscribeNewHeadsRequest(subId: String): JsonRpcRequest

-    fun localReaderBuilder(cachingReader: CachingReader, methods: CallMethods, head: Head): Mono<JsonRpcReader>
+    fun localReaderBuilder(cachingReader: CachingReader, methods: CallMethods, head: Head, logsOracle: LogsOracle?): Mono<JsonRpcReader>

    fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription

--- 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<String, Head> =
@@ -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<JsonRpcReader> {
-        return localReaderBuilder(cachingReader, getMethods(), getHead())
+        return localReaderBuilder(cachingReader, getMethods(), getHead(), logsOracle)
    }

    override fun getEgressSubscription(): EgressSubscription {
--- 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<JsonRpcReader> {
        return Mono.just(LocalReader(methods))
    }
--- a/src/main/resources/LogsOracle.jar
+++ b/src/main/resources/LogsOracle.jar