problem: bitcoin mempool can be queries too often

solution: introduce data access wrappers, in this case caching mempool state in memory
2020-04-27 00:05:02 -04:00
parent c9082b3cd0
commit bd58ba9774
15 changed files with 197 additions and 73 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
@@ -6,7 +6,7 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstreams
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -48,7 +48,7 @@ class TrackBitcoinAddress(
        }
    }

-    fun requestBalances(chain: Chain, api: BitcoinApi, addresses: List<String>): Flux<AddressBalance> {
+    fun requestBalances(chain: Chain, api: DirectBitcoinApi, addresses: List<String>): Flux<AddressBalance> {
        return api.executeAndResult(0, "listunspent", emptyList(), List::class.java)
                .flatMapMany { unspents ->
                    val result = getTotal(chain, addresses, unspents)
@@ -58,7 +58,7 @@ class TrackBitcoinAddress(

    override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
        val chain = Chain.byId(request.asset.chainValue)
-        val upstream = upstreams.getUpstream(chain)?.castApi(BitcoinApi::class.java)
+        val upstream = upstreams.getUpstream(chain)?.castApi(DirectBitcoinApi::class.java)
                ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
        val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
        if (addresses.isEmpty()) {
@@ -107,7 +107,7 @@ class TrackBitcoinAddress(

    override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
        val chain = Chain.byId(request.asset.chainValue)
-        val upstream = upstreams.getUpstream(chain)?.castApi(BitcoinApi::class.java)
+        val upstream = upstreams.getUpstream(chain)?.castApi(DirectBitcoinApi::class.java)
                ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
        val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
        if (addresses.isEmpty()) {
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt
@@ -6,9 +6,9 @@ import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.upstream.Selector
-import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.Upstreams
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
 import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
@@ -37,7 +37,7 @@ class TrackBitcoinTx(

    override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
        val chain = Chain.byId(request.chainValue)
-        val upstream = upstreams.getUpstream(chain)?.castApi(BitcoinApi::class.java)
+        val upstream = upstreams.getUpstream(chain)?.cast(BitcoinUpstream::class.java, DirectBitcoinApi::class.java)
                ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
        val txid = request.txId
        val confirmations = max(min(1, request.confirmationLimit), 12)
@@ -48,7 +48,7 @@ class TrackBitcoinTx(
        }.map(this::asProto)
    }

-    fun subscribe(chain: Chain, api: BitcoinApi, upstream: Upstream<BitcoinApi>, txid: String): Flux<TxStatus> {
+    fun subscribe(chain: Chain, api: DirectBitcoinApi, upstream: BitcoinUpstream, txid: String): Flux<TxStatus> {
        return loadExisting(api, txid)
                .flatMapMany { status ->
                    if (status.mined) {
@@ -56,7 +56,7 @@ class TrackBitcoinTx(
                        //without publishing an empty TxStatus first
                        continueWithMined(api, upstream, status)
                    } else {
-                        loadMempool(api, txid)
+                        loadMempool(upstream, txid)
                                .flatMapMany { tx ->
                                    val next = if (tx.found) {
                                        untilMined(upstream, tx)
@@ -70,7 +70,7 @@ class TrackBitcoinTx(
                }
    }

-    fun continueWithMined(api: BitcoinApi, upstream: Upstream<BitcoinApi>, status: TxStatus): Flux<TxStatus> {
+    fun continueWithMined(api: DirectBitcoinApi, upstream: BitcoinUpstream, status: TxStatus): Flux<TxStatus> {
        return api.getBlock(status.blockHash!!)
                .map { block ->
                    TxStatus(status.txid, true, ExtractBlock.getHeight(block), true, status.blockHash, ExtractBlock.getTime(block), ExtractBlock.getDifficulty(block))
@@ -79,10 +79,10 @@ class TrackBitcoinTx(
                }
    }

-    fun untilFound(chain: Chain, api: BitcoinApi, upstream: Upstream<BitcoinApi>, txid: String): Flux<TxStatus> {
+    fun untilFound(chain: Chain, api: DirectBitcoinApi, upstream: BitcoinUpstream, txid: String): Flux<TxStatus> {
        return Flux.interval(Duration.ofSeconds(1))
                .take(Duration.ofMinutes(10))
-                .flatMap { loadMempool(api, txid) }
+                .flatMap { loadMempool(upstream, txid) }
                .skipUntil { it.found }
                .flatMap { subscribe(chain, api, upstream, txid) }
                .doOnError { t ->
@@ -90,7 +90,7 @@ class TrackBitcoinTx(
                }
    }

-    fun untilMined(upstream: Upstream<BitcoinApi>, tx: TxStatus): Mono<TxStatus> {
+    fun untilMined(upstream: BitcoinUpstream, tx: TxStatus): Mono<TxStatus> {
        return upstream.getHead().getFlux().flatMap {
            upstream.getApi(Selector.empty).flatMap { api ->
                loadExisting(api, tx.txid)
@@ -98,13 +98,13 @@ class TrackBitcoinTx(
        }.single()
    }

-    fun withConfirmations(upstream: Upstream<BitcoinApi>, tx: TxStatus): Flux<TxStatus> {
+    fun withConfirmations(upstream: BitcoinUpstream, tx: TxStatus): Flux<TxStatus> {
        return upstream.getHead().getFlux().map {
            tx.withHead(it.height)
        }
    }

-    fun loadExisting(api: BitcoinApi, txid: String): Mono<TxStatus> {
+    fun loadExisting(api: DirectBitcoinApi, txid: String): Mono<TxStatus> {
        val mined = api.getTx(txid)
        return mined.map {
            val block = it["blockhash"] as String?
@@ -112,8 +112,9 @@ class TrackBitcoinTx(
        }
    }

-    fun loadMempool(api: BitcoinApi, txid: String): Mono<TxStatus> {
-        val mempool = api.getMempool()
+    fun loadMempool(upstream: BitcoinUpstream, txid: String): Mono<TxStatus> {
+        println("access: ${upstream.getData()}")
+        val mempool = upstream.getData().getMempool().get()
        return mempool.map {
            if (it.contains(txid)) {
                TxStatus(txid, found = true, mined = false)
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -20,10 +20,9 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.CurrentUpstreams
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcClient
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
-import io.emeraldpay.dshackle.upstream.bitcoin.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
@@ -134,11 +133,11 @@ open class ConfiguredUpstreams(
                                     options: UpstreamsConfig.Options) {

        val conn = config.connection!!
-        var rpcApi: BitcoinApi? = null
+        var rpcApi: DirectBitcoinApi? = null
        val methods = buildMethods(config, chain)
        conn.rpc?.let { endpoint ->
            val rpcClient = BitcoinRpcClient(endpoint.url.toString(), endpoint.basicAuth!!)
-            rpcApi = BitcoinApi(rpcClient, objectMapper, methods)
+            rpcApi = DirectBitcoinApi(rpcClient, objectMapper, methods)
        }
        rpcApi?.let { api ->
            val upstream = BitcoinUpstream(config.id
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.cache.CachesFactory
 import io.emeraldpay.dshackle.startup.UpstreamChange
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
 import io.emeraldpay.dshackle.upstream.bitcoin.DefaultBitcoinMethods
@@ -70,10 +70,10 @@ class CurrentUpstreams(
                }
                BlockchainType.BITCOIN -> {
                    val up = change.upstream
-                            .cast(BitcoinUpstream::class.java, BitcoinApi::class.java)
-                    val current = chainMapping[chain] as ChainUpstreams<BitcoinApi>?
+                            .cast(BitcoinUpstream::class.java, DirectBitcoinApi::class.java)
+                    val current = chainMapping[chain] as ChainUpstreams<DirectBitcoinApi>?
                    val factory = Callable {
-                        BitcoinChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<BitcoinApi>
+                        BitcoinChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<DirectBitcoinApi>
                    }
                    processUpdate(change, up, current, factory)
                }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt
@@ -14,7 +14,7 @@ class BitcoinChainUpstreams(
        val upstreams: MutableList<BitcoinUpstream>,
        caches: Caches,
        objectMapper: ObjectMapper
-) : ChainUpstreams<BitcoinApi>(chain, upstreams as MutableList<Upstream<BitcoinApi>>, caches, objectMapper) {
+) : ChainUpstreams<DirectBitcoinApi>(chain, upstreams as MutableList<Upstream<DirectBitcoinApi>>, caches, objectMapper) {

    companion object {
        private val log = LoggerFactory.getLogger(BitcoinChainUpstreams::class.java)
@@ -66,7 +66,7 @@ class BitcoinChainUpstreams(
    }

    override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
-        if (!apiType.isAssignableFrom(BitcoinApi::class.java)) {
+        if (!apiType.isAssignableFrom(DirectBitcoinApi::class.java)) {
            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
        }
        return this as Upstream<A>
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinData.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinData.kt
@@ -0,0 +1,33 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.upstream.Head
+import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+
+open class BitcoinData(
+        api: DirectBitcoinApi,
+        head: Head
+) : Lifecycle {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BitcoinData::class.java)
+    }
+
+    private val mempool = CachingMempoolData(api, head)
+
+    open fun getMempool(): CachingMempoolData {
+        return mempool
+    }
+
+    override fun isRunning(): Boolean {
+        return mempool.isRunning
+    }
+
+    override fun start() {
+        mempool.start()
+    }
+
+    override fun stop() {
+        mempool.stop()
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
@@ -3,7 +3,6 @@ package io.emeraldpay.dshackle.upstream.bitcoin
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.upstream.AbstractHead
 import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory
@@ -15,7 +14,7 @@ import java.time.Duration
 import java.util.concurrent.Executors

 class BitcoinRpcHead(
-        private val api: BitcoinApi,
+        private val api: DirectBitcoinApi,
        private val extractBlock: ExtractBlock,
        private val interval: Duration = Duration.ofSeconds(15)
 ) : Head, AbstractHead(), Lifecycle {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
@@ -6,21 +6,20 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.grpc.Chain
-import io.infinitape.etherjar.rpc.JacksonRpcConverter
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Mono

-class BitcoinUpstream(
+open class BitcoinUpstream(
        id: String,
        val chain: Chain,
-        private val api: BitcoinApi,
+        private val api: DirectBitcoinApi,
        options: UpstreamsConfig.Options,
        val node: QuorumForLabels.QuorumItem,
        private val objectMapper: ObjectMapper,
        callMethods: CallMethods
-) : DefaultUpstream<BitcoinApi>(id, options, callMethods), Lifecycle {
+) : DefaultUpstream<DirectBitcoinApi>(id, options, callMethods), Lifecycle {

    companion object {
        private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java)
@@ -28,6 +27,7 @@ class BitcoinUpstream(

    private val head: Head = createHead()
    private var validatorSubscription: Disposable? = null
+    private val data = BitcoinData(api, head)

    private fun createHead(): Head {
        return BitcoinRpcHead(
@@ -36,11 +36,15 @@ class BitcoinUpstream(
        )
    }

+    open fun getData(): BitcoinData {
+        return data
+    }
+
    override fun getHead(): Head {
        return head
    }

-    override fun getApi(matcher: Selector.Matcher): Mono<out BitcoinApi> {
+    override fun getApi(matcher: Selector.Matcher): Mono<out DirectBitcoinApi> {
        return Mono.just(api)
    }

@@ -56,8 +60,8 @@ class BitcoinUpstream(
    }

    override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
-        if (!apiType.isAssignableFrom(BitcoinApi::class.java)) {
-            throw ClassCastException("Cannot cast ${BitcoinApi::class.java} to $apiType")
+        if (!apiType.isAssignableFrom(DirectBitcoinApi::class.java)) {
+            throw ClassCastException("Cannot cast ${DirectBitcoinApi::class.java} to $apiType")
        }
        return this as Upstream<A>
    }
@@ -67,6 +71,7 @@ class BitcoinUpstream(
        if (head is Lifecycle) {
            runningAny = runningAny || head.isRunning
        }
+        runningAny = runningAny || data.isRunning
        return runningAny
    }

@@ -77,6 +82,8 @@ class BitcoinUpstream(
                head.start()
            }
        }
+        data.start()
+
        validatorSubscription?.dispose()

        if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
@@ -93,6 +100,7 @@ class BitcoinUpstream(
        if (head is Lifecycle) {
            head.stop()
        }
+        data.stop()
        validatorSubscription?.dispose()
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt
@@ -1,10 +1,7 @@
 package io.emeraldpay.dshackle.upstream.bitcoin

-import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
-import io.infinitape.etherjar.rpc.Commands
 import org.slf4j.LoggerFactory
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory
 import reactor.core.publisher.Flux
@@ -14,7 +11,7 @@ import java.time.Duration
 import java.util.concurrent.Executors

 class BitcoinUpstreamValidator(
-        private val api: BitcoinApi,
+        private val api: DirectBitcoinApi,
        private val options: UpstreamsConfig.Options
 ) {

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt
@@ -0,0 +1,71 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.upstream.Head
+import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
+import reactor.core.publisher.Mono
+import java.time.Duration
+import java.time.Instant
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.locks.ReentrantLock
+
+open class CachingMempoolData(
+        private val api: DirectBitcoinApi,
+        private val head: Head
+) : Lifecycle {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(CachingMempoolData::class.java)
+        private val TTL = Duration.ofSeconds(15)
+    }
+
+    private val current = AtomicReference<Container>(Container.empty())
+    private val updateLock = ReentrantLock()
+    private var headListener: Disposable? = null
+
+    open fun get(): Mono<List<String>> {
+        val value = current.get()
+        return if (value.since < Instant.now().minus(TTL)) {
+            updateLock.lock()
+            fetchFromUpstream()
+                    .timeout(Duration.ofSeconds(3), Mono.empty())
+                    .doOnNext {
+                        current.set(Container(Instant.now(), it))
+                    }.doFinally {
+                        updateLock.unlock()
+                    }
+        } else {
+            Mono.just(value.value)
+        }
+    }
+
+    fun fetchFromUpstream(): Mono<List<String>> {
+        return api.executeAndResult(0, "getrawmempool", emptyList(), List::class.java) as Mono<List<String>>
+    }
+
+    class Container(val since: Instant, val value: List<String>) {
+        companion object {
+            fun empty(): Container {
+                return Container(Instant.MIN, emptyList())
+            }
+        }
+    }
+
+    override fun isRunning(): Boolean {
+        return headListener != null
+    }
+
+    override fun start() {
+        headListener?.dispose()
+        headListener = head.getFlux().doOnNext {
+            current.set(Container.empty())
+        }.subscribe()
+    }
+
+    override fun stop() {
+        headListener?.dispose()
+        headListener = null
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/DirectBitcoinApi.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/DirectBitcoinApi.kt
@@ -14,14 +14,14 @@ import io.infinitape.etherjar.rpc.json.ResponseJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono

-open class BitcoinApi(
+open class DirectBitcoinApi(
        val bitcoinRpcClient: BitcoinRpcClient,
        val objectMapper: ObjectMapper,
        val targets: CallMethods
 ) : UpstreamApi {

    companion object {
-        private val log = LoggerFactory.getLogger(BitcoinApi::class.java)
+        private val log = LoggerFactory.getLogger(DirectBitcoinApi::class.java)
    }

    open override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
@@ -104,7 +104,4 @@ open class BitcoinApi(
        return executeAndResult(0, "getrawtransaction", listOf(txid, true), Map::class.java) as Mono<Map<String, Any>>
    }

-    open fun getMempool(): Mono<List<String>> {
-        return executeAndResult(0, "getrawmempool", emptyList(), List::class.java) as Mono<List<String>>
-    }
 }