Merge pull request #265 from emeraldpay/fix/no-btc-balance

problem: doesn't fetch Bitcoin balance when provided by another Dshac…
emeraldpay · Oct 10, 2023 · 71bf277 · 71bf277
2 parents 714ea1d + cbff4a9
commit 71bf277
Show file tree

Hide file tree

Showing 4 changed files with 102 additions and 9 deletions.
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
@@ -78,7 +78,7 @@ class TrackBitcoinAddress(
         multistreamHolder.observeChains().subscribe { chain ->
             multistreamHolder.getUpstream(chain)?.let { mup ->
                 val available = mup.getAll().any { up ->
-                    !up.isGrpc() && up.getCapabilities().contains(Capability.BALANCE)
+                    up.getCapabilities().contains(Capability.BALANCE)
                 }
                 setBalanceAvailability(chain, available)
             }

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
@@ -86,6 +86,12 @@ abstract class DefaultUpstream(
             throw IllegalArgumentException("Invalid upstream id: $id")
         }
 
+        if (options.disableValidation) {
+            // if we specifically told that this upstream should be _always valid_ start with this state,
+            // but note it could be updated later (ex. provided by gRPC upstream)
+            this.setStatus(UpstreamAvailability.OK)
+        }
+
         forkWatch.register(this)
             .subscribeOn(Schedulers.boundedElastic())
             .subscribe {

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReader.kt
@@ -1,6 +1,10 @@
 package io.emeraldpay.dshackle.upstream.bitcoin
 
 import io.emeraldpay.api.proto.BlockchainOuterClass.BalanceRequest
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.api.proto.Common.AnyAddress
+import io.emeraldpay.api.proto.Common.SingleAddress
+import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
@@ -26,14 +30,15 @@ class RemoteUnspentReader(
         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 { up -> up.cast(BitcoinGrpcUpstream::class.java) }
+            .flatMap { readFromUpstream(it, key) }
+    }
+
+    fun readFromUpstream(upstream: BitcoinGrpcUpstream, address: Address): Mono<List<SimpleUnspent>> {
+        val request = createRequest(address)
+        return upstream.remote.getBalance(request)
+            .switchIfEmpty(Mono.error(SilentException.DataUnavailable("Balance not provider")))
+            .doOnError { t -> log.warn("Failed to get balance from remote", t) }
             .map { resp ->
                 resp.utxoList.map { utxo ->
                     SimpleUnspent(
@@ -45,4 +50,12 @@ class RemoteUnspentReader(
             }
             .reduce(List<SimpleUnspent>::plus)
     }
+
+    fun createRequest(address: Address): BalanceRequest {
+        return BalanceRequest.newBuilder()
+            .setAddress(AnyAddress.newBuilder().setAddressSingle(SingleAddress.newBuilder().setAddress(address.toString())))
+            .setAsset(Common.Asset.newBuilder().setChainValue(upstreams.chain.id))
+            .setIncludeUtxo(true)
+            .build()
+    }
 }
diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReaderTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RemoteUnspentReaderTest.kt
@@ -0,0 +1,74 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.api.Chain
+import io.emeraldpay.api.proto.BlockchainOuterClass.AddressBalance
+import io.emeraldpay.api.proto.BlockchainOuterClass.BalanceRequest
+import io.emeraldpay.api.proto.BlockchainOuterClass.Utxo
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.upstream.grpc.BitcoinGrpcUpstream
+import io.kotest.core.spec.style.ShouldSpec
+import io.kotest.matchers.collections.shouldHaveSize
+import io.kotest.matchers.shouldBe
+import io.mockk.every
+import io.mockk.mockk
+import org.bitcoinj.core.Address
+import org.bitcoinj.params.MainNetParams
+import reactor.core.publisher.Flux
+import java.time.Duration
+
+class RemoteUnspentReaderTest : ShouldSpec({
+
+    should("Create a request") {
+        val ups = mockk<BitcoinMultistream>()
+        every { ups.chain } returns Chain.BITCOIN
+        val reader = RemoteUnspentReader(ups)
+
+        val act = reader.createRequest(Address.fromString(MainNetParams.get(), "38XnPvu9PmonFU9WouPXUjYbW91wa5MerL"))
+
+        act.asset.chain.name shouldBe "CHAIN_BITCOIN"
+        act.includeUtxo shouldBe true
+        act.address.addrTypeCase shouldBe Common.AnyAddress.AddrTypeCase.ADDRESS_SINGLE
+        act.address.addressSingle.address shouldBe "38XnPvu9PmonFU9WouPXUjYbW91wa5MerL"
+    }
+
+    should("Read from remote") {
+        val ups = mockk<BitcoinMultistream>()
+        every { ups.chain } returns Chain.BITCOIN
+        val reader = RemoteUnspentReader(ups)
+
+        val up = mockk<BitcoinGrpcUpstream>()
+        every { up.remote } returns mockk() {
+            every { getBalance(any() as BalanceRequest) } returns Flux.fromIterable(
+                listOf(
+                    AddressBalance.newBuilder()
+                        .addAllUtxo(
+                            listOf(
+                                Utxo.newBuilder()
+                                    .setTxId("cb46a01a257194ecf7d6a1f7e1bee8ac4f0a6687ec13bb0bba8942377b64a6c4")
+                                    .setIndex(0)
+                                    .setBalance("102030")
+                                    .build(),
+                                Utxo.newBuilder()
+                                    .setTxId("c742a3e4257194ecf7d6a1f7e1bee8ac4b066a51ec13bb0bba8942377b64a6c4")
+                                    .setIndex(1)
+                                    .setBalance("123")
+                                    .build()
+                            )
+                        )
+                        .build()
+                )
+            )
+        }
+
+        val act = reader.readFromUpstream(up, Address.fromString(MainNetParams.get(), "38XnPvu9PmonFU9WouPXUjYbW91wa5MerL"))
+            .block(Duration.ofSeconds(1))
+
+        act shouldHaveSize 2
+        act[0].txid shouldBe "cb46a01a257194ecf7d6a1f7e1bee8ac4f0a6687ec13bb0bba8942377b64a6c4"
+        act[0].vout shouldBe 0
+        act[0].value shouldBe 102030
+        act[1].txid shouldBe "c742a3e4257194ecf7d6a1f7e1bee8ac4b066a51ec13bb0bba8942377b64a6c4"
+        act[1].vout shouldBe 1
+        act[1].value shouldBe 123
+    }
+})