problem: Bitcoin listunspent is not working when the upstream do not provide it, though have other sources of unspent data

solution: wrapper for listunspent requests that uses Dshackle mechanism to gather data
2022-07-21 22:09:37 -04:00
parent 7f67171ad6
commit d65a217535
16 changed files with 144 additions and 41 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
@@ -180,8 +180,8 @@ class TrackBitcoinAddress(
            .timeout(Defaults.timeoutInternal, Mono.empty())
            .switchIfEmpty(
                Mono.fromCallable {
-                        log.warn("No upstream providing balance for ${api.chain}")
-                    }
+                    log.warn("No upstream providing balance for ${api.chain}")
+                }
                    .then(Mono.error(SilentException.DataUnavailable("BALANCE")))
            )
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
@@ -26,7 +26,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.RequestPostprocessor
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.ethereum.LocalCallRouter
+import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
@@ -37,28 +37,34 @@ import reactor.core.publisher.Mono
@Suppress("UNCHECKED_CAST")
 open class BitcoinMultistream(
    chain: Chain,
-    val upstreams: MutableList<BitcoinUpstream>,
+    private val sourceUpstreams: MutableList<BitcoinUpstream>,
    caches: Caches
-) : Multistream(chain, upstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty()), Lifecycle {
+) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty()), Lifecycle {

    companion object {
        private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java)
    }

    private var head: Head = EmptyHead()
-    private var esplora = upstreams.find { it.esploraClient != null }?.esploraClient
+    private var esplora = sourceUpstreams.find { it.esploraClient != null }?.esploraClient
    private var reader = BitcoinReader(this, head, esplora)
    private var addressActiveCheck: AddressActiveCheck? = null
    private var xpubAddresses: XpubAddresses? = null
    private val feeEstimation = BitcoinFees(this, reader, 6)
+    private var callRouter: LocalCallRouter = LocalCallRouter(DefaultBitcoinMethods(), reader)

    override fun init() {
-        if (upstreams.size > 0) {
+        if (sourceUpstreams.size > 0) {
            head = updateHead()
        }
        super.init()
    }

+    open val upstreams: List<BitcoinUpstream>
+        get() {
+            return sourceUpstreams
+        }
+
    override fun getFeeEstimation(): ChainFees {
        return feeEstimation
    }
@@ -75,8 +81,8 @@ open class BitcoinMultistream(
        }
        lagObserver?.stop()
        lagObserver = null
-        val head = if (upstreams.size == 1) {
-            val upstream = upstreams.first()
+        val head = if (sourceUpstreams.size == 1) {
+            val upstream = sourceUpstreams.first()
            upstream.setLag(0)
            upstream.getHead().apply {
                if (this is Lifecycle) {
@@ -84,10 +90,10 @@ open class BitcoinMultistream(
                }
            }
        } else {
-            val newHead = MergedHead(upstreams.map { it.getHead() }).apply {
+            val newHead = MergedHead(sourceUpstreams.map { it.getHead() }).apply {
                this.start()
            }
-            val lagObserver = BitcoinHeadLagObserver(newHead, upstreams)
+            val lagObserver = BitcoinHeadLagObserver(newHead, sourceUpstreams)
            this.lagObserver = lagObserver
            lagObserver.start()
            newHead
@@ -97,7 +103,7 @@ open class BitcoinMultistream(
    }

    override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
-        return Mono.just(LocalCallRouter(getMethods()))
+        return Mono.just(callRouter)
    }

    open fun getReader(): BitcoinReader {
@@ -106,10 +112,11 @@ open class BitcoinMultistream(

    override fun onUpstreamsUpdated() {
        super.onUpstreamsUpdated()
-        esplora = upstreams.find { it.esploraClient != null }?.esploraClient
+        esplora = sourceUpstreams.find { it.esploraClient != null }?.esploraClient
        reader = BitcoinReader(this, this.head, esplora)
        addressActiveCheck = esplora?.let { AddressActiveCheck(it) }
        xpubAddresses = addressActiveCheck?.let { XpubAddresses(it) }
+        callRouter = LocalCallRouter(getMethods(), reader)
    }

    override fun setHead(head: Head) {
@@ -122,7 +129,7 @@ open class BitcoinMultistream(
    }

    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
-        return upstreams.flatMap { it.getLabels() }
+        return sourceUpstreams.flatMap { it.getLabels() }
    }

    @Suppress("UNCHECKED_CAST")
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin

 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
@@ -43,6 +44,8 @@ open class BitcoinReader(

    private val unspentReader: UnspentReader = if (esploraClient != null) {
        EsploraUnspentReader(esploraClient, head)
+    } else if (upstreams.upstreams.any { it.isGrpc() && it.getCapabilities().contains(Capability.BALANCE) }) {
+        RemoteUnspentReader(upstreams)
    } else {
        RpcUnspentReader(upstreams)
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraUnspentReader.kt
@@ -37,7 +37,6 @@ class EsploraUnspentReader(
            base.txid,
            base.vout,
            base.value,
-            head.getCurrentHeight()?.let { base.height - it } ?: 0
        )
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt
@@ -15,12 +15,17 @@
 */
 package io.emeraldpay.dshackle.upstream.bitcoin

+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
+import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
+import org.bitcoinj.core.Address
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono

@@ -33,6 +38,7 @@ import reactor.core.publisher.Mono
 */
 class LocalCallRouter(
    private val methods: CallMethods,
+    private val reader: BitcoinReader,
 ) : Reader<JsonRpcRequest, JsonRpcResponse> {

    companion object {
@@ -47,6 +53,39 @@ class LocalCallRouter(
        if (!methods.isCallable(key.method)) {
            return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
        }
+        if (key.method == "listunspent") {
+            return processUnspentRequest(key)
+        }
        return Mono.empty()
    }
+
+    /**
+     *
+     */
+    fun processUnspentRequest(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+        if (key.params.size < 3) {
+            return Mono.error(SilentException("Invalid call to unspent. Address is missing"))
+        }
+        val addresses = key.params[2]
+        if (addresses is List<*> && addresses.size > 0) {
+            val address = addresses[0].toString().let { Address.fromString(null, it) }
+            return reader.listUnspent(address).map {
+                val rpc = it.map(convertUnspent(address))
+                val json = Global.objectMapper.writeValueAsBytes(rpc)
+                JsonRpcResponse.ok(json, JsonRpcResponse.NumberId(key.id))
+            }
+        }
+        return Mono.error(SilentException("Invalid call to unspent"))
+    }
+
+    fun convertUnspent(address: Address): (SimpleUnspent) -> RpcUnspent {
+        return { base ->
+            RpcUnspent(
+                base.txid,
+                base.vout,
+                address.toString(),
+                base.value,
+            )
+        }
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt
@@ -0,0 +1,48 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.api.proto.BlockchainOuterClass.BalanceRequest
+import io.emeraldpay.dshackle.upstream.Capability
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
+import io.emeraldpay.dshackle.upstream.grpc.BitcoinGrpcUpstream
+import org.bitcoinj.core.Address
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+class RemoteUnspentReader(
+    val upstreams: BitcoinMultistream
+) : UnspentReader {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(RemoteUnspentReader::class.java)
+    }
+
+    private val selector = Selector.LocalAndMatcher(
+        Selector.GrpcMatcher(),
+        Selector.CapabilityMatcher(Capability.BALANCE)
+    )
+
+    override fun read(key: Address): Mono<List<SimpleUnspent>> {
+        val apis = upstreams.getApiSource(selector)
+        apis.request(1)
+        return Mono.from(apis)
+            .map { up ->
+                up.cast(BitcoinGrpcUpstream::class.java).remote
+            }
+            .flatMapMany {
+                val request = BalanceRequest.newBuilder()
+                    .build()
+                it.getBalance(request)
+            }
+            .map { resp ->
+                resp.utxoList.map { utxo ->
+                    SimpleUnspent(
+                        utxo.txId,
+                        utxo.index.toInt(),
+                        utxo.balance.toLong(),
+                    )
+                }
+            }
+            .reduce(List<SimpleUnspent>::plus)
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt
@@ -16,6 +16,8 @@
 package io.emeraldpay.dshackle.upstream.bitcoin

 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.SilentException
+import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
 import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
@@ -38,14 +40,17 @@ class RpcUnspentReader(
            base.txid,
            base.vout,
            base.amount,
-            base.confirmations
        )
    }

+    private val selector = Selector.CapabilityMatcher(Capability.BALANCE)
+
    override fun read(key: Address): Mono<List<SimpleUnspent>> {
+        // docs: https://developer.bitcoin.org/reference/rpc/listunspent.html
+        //
        val address = key.toString()
-        return upstreams.getDirectApi(Selector.empty).flatMap { api ->
-            api.read(JsonRpcRequest("listunspent", emptyList()))
+        return upstreams.getDirectApi(selector).flatMap { api ->
+            api.read(JsonRpcRequest("listunspent", listOf(1, 9999999, listOf(address))))
                .flatMap(JsonRpcResponse::requireResult)
                .map {
                    Global.objectMapper.readerFor(RpcUnspent::class.java).readValues<RpcUnspent>(it).readAll()
@@ -55,6 +60,6 @@ class RpcUnspentReader(
                        it.address == address
                    }.map(convert)
                }
-        }
+        }.switchIfEmpty(Mono.error(SilentException.DataUnavailable("BALANCE")))
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspent.kt
@@ -20,5 +20,4 @@ data class RpcUnspent(
    val vout: Int,
    val address: String,
    val amount: Long,
-    val confirmations: Long
 )
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/RpcUnspentDeserializer.kt
@@ -30,7 +30,6 @@ class RpcUnspentDeserializer : JsonDeserializer<RpcUnspent>() {
            node.get("vout").asInt(),
            node.get("address").asText(),
            BigDecimal(node.get("amount").asText()).multiply(BigDecimal.TEN.pow(8)).longValueExact(),
-            node.get("confirmations").asLong()
        )
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/data/SimpleUnspent.kt
@@ -19,5 +19,4 @@ data class SimpleUnspent(
    val txid: String,
    val vout: Int,
    val value: Long,
-    val confirmations: Long
 )
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
@@ -99,6 +99,10 @@ class BitcoinGrpcUpstream(
    var timeout = Defaults.timeout
    private var capabilities: Set<Capability> = emptySet()

+    override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub {
+        return remote
+    }
+
    override fun getHead(): Head {
        return grpcHead
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
@@ -103,6 +103,10 @@ open class EthereumGrpcUpstream(
    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)
    var timeout = Defaults.timeout

+    override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub {
+        return remote
+    }
+
    override fun start() {
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt
@@ -16,12 +16,16 @@
 package io.emeraldpay.dshackle.upstream.grpc

 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.dshackle.upstream.Upstream

-interface GrpcUpstream {
+interface GrpcUpstream : Upstream {

    /**
     * Update the configuration of the upstream with the new data.
     * Called on the first creation, and each time a new state received from upstream
     */
    fun update(conf: BlockchainOuterClass.DescribeChain)
+
+    fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub
 }