From d537ff046192d0bc2dc3a5d4ee2e878d04df3d6e Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 22 Sep 2020 23:51:07 -0400 Subject: [PATCH] solution: delegate bitcoin balance request to another upstream, if data is unavailable locally --- docs/04-upstream-config.adoc | 1 + docs/reference-configuration.adoc | 6 + .../io/emeraldpay/dshackle/SilentException.kt | 4 +- .../dshackle/config/UpstreamsConfig.kt | 2 + .../dshackle/config/UpstreamsConfigReader.kt | 3 + .../io/emeraldpay/dshackle/rpc/Describe.kt | 11 +- .../dshackle/rpc/TrackBitcoinAddress.kt | 120 +++++++++++++++--- .../upstream/CurrentMultistreamHolder.kt | 2 +- .../dshackle/upstream/Multistream.kt | 19 ++- .../emeraldpay/dshackle/upstream/Selector.kt | 25 +++- .../emeraldpay/dshackle/upstream/Upstream.kt | 2 + .../upstream/bitcoin/BitcoinRpcUpstream.kt | 14 ++ .../upstream/ethereum/EthereumRpcUpstream.kt | 13 ++ .../upstream/grpc/BitcoinGrpcUpstream.kt | 17 ++- .../upstream/grpc/EthereumGrpcUpstream.kt | 9 ++ .../upstream/grpc/RemoteCapabilities.kt | 23 ++++ .../config/UpstreamsConfigReaderSpec.groovy | 2 + .../rpc/TrackBitcoinAddressSpec.groovy | 2 +- .../grpc/RemoteCapabilitiesSpec.groovy | 32 +++++ .../resources/upstreams-bitcoin-esplora.yaml | 2 + 20 files changed, 276 insertions(+), 33 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilitiesSpec.groovy diff --git a/docs/04-upstream-config.adoc b/docs/04-upstream-config.adoc index e672b61f..a7c7f295 100644 --- a/docs/04-upstream-config.adoc +++ b/docs/04-upstream-config.adoc @@ -125,6 +125,7 @@ Options (default or as part of upstream config): provider such as Infura, but disabling it is not recommended for a normal node) | `min-peers` | 3 | specify minimum amount of connected peers, Dshackle will not use upstream with less than specified number | `timeout` | 60 | timeout in seconds after which request to the upstream will be discarded (and may be retried on an another upstream) +| `balance` | `true` for ethereum, `false` for bitcoin | specify if this node should be used to fetch balance for an address |=== === Connection type diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index e7f4084c..35f618b9 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -103,6 +103,9 @@ cluster: password: 1a68f20154fc258fe4149c199ad8f281 - id: bitcoin chain: bitcoin + options: + # use the node to fetch balances + balance: true connection: bitcoin: rpc: @@ -110,6 +113,9 @@ cluster: basic-auth: username: bitcoin password: e984af45bb888428207c290 + # uses Esplora index to fetch balances and utxo for an address + esplora: + url: "http://localhost:3001" - id: remote connection: grpc: diff --git a/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt b/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt index c8c3513b..5fe77aa7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt @@ -26,7 +26,9 @@ open class SilentException(message: String) : Exception(message) { /** * Blockchain is not available or not supported by current instance of the Dshackle */ - class UnsupportedBlockchain(val blockchainId: Int): SilentException("Unsupported blockchain $blockchainId") { + class UnsupportedBlockchain(val blockchainId: Int) : SilentException("Unsupported blockchain $blockchainId") { constructor(chain: Chain) : this(chain.id) } + + class DataUnavailable(val code: String) : SilentException("Data is unavailable: $code") } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 70881287..e9a50e09 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -30,6 +30,7 @@ class UpstreamsConfig { open class Options { var disableValidation: Boolean? = null var timeout = Defaults.timeout + var providesBalance: Boolean? = null var minPeers: Int? = 1 set(minPeers) { @@ -46,6 +47,7 @@ class UpstreamsConfig { val copy = Options() copy.minPeers = if (this.minPeers != null) this.minPeers else additional.minPeers copy.disableValidation = if (this.disableValidation != null) this.disableValidation else additional.disableValidation + copy.providesBalance = if (this.providesBalance != null) this.providesBalance else additional.providesBalance return copy } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 8ded3edb..d716a3cf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -261,6 +261,9 @@ class UpstreamsConfigReader( getValueAsBool(values, "disable-validation")?.let { options.disableValidation = it } + getValueAsBool(values, "balance")?.let { + options.providesBalance = it + } return options } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index afe64fc6..3e995499 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -20,7 +20,6 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.* -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import reactor.core.publisher.Mono @@ -38,6 +37,7 @@ class Describe( multistreamHolder.getUpstream(chain)?.let { chainUpstreams -> val status = subscribeStatus.chainStatus(chain, chainUpstreams.getAll()) val targets = chainUpstreams.getMethods().getSupportedMethods() + val capabilities: MutableSet = mutableSetOf() val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder() .setChain(Common.ChainRef.forNumber(chain.id)) .addAllSupportedMethods(targets) @@ -59,8 +59,17 @@ class Describe( }) chainDescription.addNodes(nodeDetails) } + capabilities.addAll(up.getCapabilities()) } } + chainDescription.addAllCapabilities( + capabilities.map { + when (it) { + Capability.RPC -> BlockchainOuterClass.Capabilities.CAP_CALLS + Capability.BALANCE -> BlockchainOuterClass.Capabilities.CAP_BALANCE + } + } + ) resp.addChains(chainDescription.build()) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt index 43167e8b..d4b80e34 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt @@ -17,11 +17,16 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common +import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.BlockchainType +import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.SilentException +import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent +import io.emeraldpay.dshackle.upstream.grpc.BitcoinGrpcUpstream import io.emeraldpay.grpc.Chain import org.bitcoinj.params.MainNetParams import org.bitcoinj.params.TestNet3Params @@ -31,6 +36,9 @@ import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger +import java.time.Duration +import java.util.concurrent.ConcurrentHashMap +import javax.annotation.PostConstruct import kotlin.collections.HashMap @Service @@ -47,6 +55,42 @@ class TrackBitcoinAddress( && BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain) } + /** + * Keep tracking of the current state of local upstreams. True for a chain that has an upstream with balance data. + */ + private val balanceAvailable: MutableMap = ConcurrentHashMap() + + /** + * Criteria for a remote grpc upstream that can provide a balance + */ + private val balanceUpstreamMatcher = Selector.LocalAndMatcher( + Selector.GrpcMatcher(), + Selector.CapabilityMatcher(Capability.BALANCE) + ) + + @PostConstruct + fun listenChains() { + multistreamHolder.observeChains().subscribe { chain -> + multistreamHolder.getUpstream(chain)?.let { mup -> + val available = mup.getAll().any { up -> + !up.isGrpc() && (up.getOptions().providesBalance ?: false) + } + setBalanceAvailability(chain, available) + } + } + } + + fun setBalanceAvailability(chain: Chain, enabled: Boolean) { + balanceAvailable[chain] = enabled + } + + /** + * @return true if the current instance has data sources to provide the balance + */ + fun isBalanceAvailable(chain: Chain): Boolean { + return balanceAvailable[chain] ?: false + } + fun allAddresses(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux { if (!request.hasAddress()) { return Flux.empty() @@ -117,38 +161,74 @@ class TrackBitcoinAddress( } } + fun getBalanceGrpc(api: BitcoinMultistream): Mono { + val ups = api.getApiSource(balanceUpstreamMatcher) + ups.request(1) + return Mono.from(ups) + .map { up -> + up.cast(BitcoinGrpcUpstream::class.java).remote + } + .timeout(Defaults.timeoutInternal, Mono.empty()) + .switchIfEmpty( + Mono.just(0) + .doOnNext { + log.warn("No upstream providing balance for ${api.chain}") + } + .then(Mono.error(SilentException.DataUnavailable("BALANCE"))) + ) + } + + fun getRemoteBalance(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux { + return getBalanceGrpc(api).flatMapMany { remote -> + remote.getBalance(request) + } + } + + fun subscribeRemoteBalance(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux { + return getBalanceGrpc(api).flatMapMany { remote -> + remote.subscribeBalance(request) + } + } + override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java) ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) - val addresses = allAddresses(upstream, request) ?: return Flux.error(SilentException("Unsupported address")) - return requestBalances(chain, upstream, addresses, request.includeUtxo) - .map(this@TrackBitcoinAddress::buildResponse) + return if (isBalanceAvailable(chain)) { + val addresses = allAddresses(upstream, request) + requestBalances(chain, upstream, addresses, request.includeUtxo) + .map(this@TrackBitcoinAddress::buildResponse) + } else { + getRemoteBalance(upstream, request) + } } - override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java) ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) - val addresses = allAddresses(upstream, request).cache() - val following = upstream.getHead().getFlux() - .flatMap { block -> - requestBalances(chain, upstream, Flux.from(addresses), request.includeUtxo) - } - val last = HashMap() - val result = following - .filter { curr -> - val prev = last[curr.address.address] - //TODO utxo can change without changing balance - val changed = prev == null || curr.balance != prev - if (changed) { - last[curr.address.address] = curr.balance + if (isBalanceAvailable(chain)) { + val addresses = allAddresses(upstream, request).cache() + val following = upstream.getHead().getFlux() + .flatMap { block -> + requestBalances(chain, upstream, Flux.from(addresses), request.includeUtxo) + } + val last = HashMap() + val result = following + .filter { curr -> + val prev = last[curr.address.address] + //TODO utxo can change without changing balance + val changed = prev == null || curr.balance != prev + if (changed) { + last[curr.address.address] = curr.balance + } + changed } - changed - } - return result.map(this@TrackBitcoinAddress::buildResponse) + return result.map(this@TrackBitcoinAddress::buildResponse) + } else { + return subscribeRemoteBalance(upstream, request) + } } private fun buildResponse(address: AddressBalance): BlockchainOuterClass.AddressBalance { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index db8c8c17..84796891 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt @@ -127,7 +127,7 @@ class CurrentMultistreamHolder( } override fun observeChains(): Flux { - return Flux.merge( + return Flux.concat( Flux.fromIterable(getAvailable()), Flux.from(chainsBus) ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 96b03e8e..8f35b2b0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -61,6 +61,7 @@ abstract class Multistream( private var seq = 0 protected var lagObserver: HeadLagObserver? = null private var subscription: Disposable? = null + private var capabilities: Set = emptySet() open fun init() { onUpstreamsUpdated() @@ -122,10 +123,18 @@ abstract class Multistream( open fun onUpstreamsUpdated() { reconfigLock.withLock { - getAll().map { it.getMethods() }.let { + val upstreams = getAll() + upstreams.map { it.getMethods() }.let { //TODO made list of uniq instances, and then if only one, just use it directly callMethods = AggregatedCallMethods(it) } + capabilities = if (upstreams.isEmpty()) { + emptySet() + } else { + upstreams.map { up -> + up.getCapabilities() + }.reduce { acc, curr -> acc + curr } + } } } @@ -211,6 +220,14 @@ abstract class Multistream( return 0 } + override fun getCapabilities(): Set { + return this.capabilities + } + + override fun isGrpc(): Boolean { + return false + } + fun printStatus() { var height: Long? = null try { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index 1e309031..1d88ecba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt @@ -149,10 +149,18 @@ class Selector { } } - class LabelMatcher(val name: String, val values: Collection): LabelSelectorMatcher() { + class LocalAndMatcher(vararg val matchers: Matcher) : Matcher { + + override fun matches(up: Upstream): Boolean { + return matchers.all { it.matches(up) } + } + + } + + class LabelMatcher(val name: String, val values: Collection) : LabelSelectorMatcher() { override fun matches(labels: UpstreamsConfig.Labels): Boolean { - return labels.get(name)?.let { - labelValue -> values.any { it == labelValue } + return labels.get(name)?.let { labelValue -> + values.any { it == labelValue } } ?: false } @@ -221,4 +229,15 @@ class Selector { } } + class CapabilityMatcher(val capability: Capability) : Matcher { + override fun matches(up: Upstream): Boolean { + return up.getCapabilities().contains(capability) + } + } + + class GrpcMatcher() : Matcher { + override fun matches(up: Upstream): Boolean { + return up.isGrpc() + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 0dce5914..5ea8fcbe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -37,6 +37,8 @@ interface Upstream { fun getLabels(): Collection fun getMethods(): CallMethods fun getId(): String + fun getCapabilities(): Set + fun isGrpc(): Boolean fun cast(selfType: Class): T } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt index d875970e..f3b5ef87 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt @@ -45,6 +45,12 @@ open class BitcoinRpcUpstream( private val head: Head = createHead() private var validatorSubscription: Disposable? = null + private val capabilities = if (options.providesBalance == true) { + setOf(Capability.RPC, Capability.BALANCE) + } else { + setOf(Capability.RPC) + } + private fun createHead(): Head { return BitcoinRpcHead( directApi, @@ -64,6 +70,14 @@ open class BitcoinRpcUpstream( return listOf(UpstreamsConfig.Labels()) } + override fun getCapabilities(): Set { + return capabilities; + } + + override fun isGrpc(): Boolean { + return false + } + override fun cast(selfType: Class): T { if (!selfType.isAssignableFrom(this.javaClass)) { throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 39e72c6e..3258a9e6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -38,6 +38,11 @@ open class EthereumRpcUpstream( private val head: Head = this.createHead() private var validatorSubscription: Disposable? = null + private val capabilities = if (options.providesBalance != false) { + setOf(Capability.RPC, Capability.BALANCE) + } else { + setOf(Capability.RPC) + } override fun setCaches(caches: Caches) { if (head is CachesEnabled) { @@ -107,6 +112,14 @@ open class EthereumRpcUpstream( return listOf(node.labels) } + override fun getCapabilities(): Set { + return capabilities + } + + override fun isGrpc(): Boolean { + return false + } + @Suppress("unchecked") override fun cast(selfType: Class): T { if (!selfType.isAssignableFrom(this.javaClass)) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 44a936d4..a9e24d8a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -22,10 +22,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient @@ -45,7 +42,7 @@ import java.util.function.Function class BitcoinGrpcUpstream( private val parentId: String, chain: Chain, - private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, + val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, private val client: JsonRpcGrpcClient ) : BitcoinUpstream( "$parentId/${chain.chainCode}", @@ -92,6 +89,7 @@ class BitcoinGrpcUpstream( private val upstreamStatus = GrpcUpstreamStatus() private val grpcHead = GrpcHead(chain, this, blockConverter, reloadBlock) var timeout = Defaults.timeout + private var capabilities: Set = emptySet() override fun getHead(): Head { return grpcHead @@ -105,6 +103,14 @@ class BitcoinGrpcUpstream( return upstreamStatus.getLabels() } + override fun getCapabilities(): Set { + return capabilities + } + + override fun isGrpc(): Boolean { + return true + } + override fun cast(selfType: Class): T { if (!selfType.isAssignableFrom(this.javaClass)) { throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") @@ -126,6 +132,7 @@ class BitcoinGrpcUpstream( override fun update(conf: BlockchainOuterClass.DescribeChain) { upstreamStatus.update(conf) + this.capabilities = RemoteCapabilities.extract(conf) conf.status?.let { status -> onStatus(status) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 2cbb7956..595b2865 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -90,6 +90,7 @@ open class EthereumGrpcUpstream( private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) private val upstreamStatus = GrpcUpstreamStatus() private val grpcHead = GrpcHead(chain, this, blockConverter, reloadBlock) + private var capabilities: Set = emptySet() private val defaultReader: Reader = client.forSelector(Selector.empty) var timeout = Defaults.timeout @@ -109,6 +110,7 @@ open class EthereumGrpcUpstream( override fun update(conf: BlockchainOuterClass.DescribeChain) { upstreamStatus.update(conf) + capabilities = RemoteCapabilities.extract(conf) conf.status?.let { status -> onStatus(status) } } @@ -148,4 +150,11 @@ open class EthereumGrpcUpstream( return this as T } + override fun getCapabilities(): Set { + return capabilities + } + + override fun isGrpc(): Boolean { + return true + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt new file mode 100644 index 00000000..3f03e14b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt @@ -0,0 +1,23 @@ +package io.emeraldpay.dshackle.upstream.grpc + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.upstream.Capability + +class RemoteCapabilities { + + companion object { + @JvmStatic + fun extract(conf: BlockchainOuterClass.DescribeChain): Set { + return conf.capabilitiesList?.let { values -> + values.mapNotNull { value -> + when { + BlockchainOuterClass.Capabilities.CAP_BALANCE == value -> Capability.BALANCE + BlockchainOuterClass.Capabilities.CAP_CALLS == value -> Capability.RPC + else -> null + } + }.toSet() + } ?: emptySet() + } + } + +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index cc06cc8c..b03526de 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -92,6 +92,7 @@ class UpstreamsConfigReaderSpec extends Specification { with(act.upstreams.get(0)) { id == "local" chain == "bitcoin" + options == null || options.providesBalance == false connection instanceof UpstreamsConfig.BitcoinConnection with((UpstreamsConfig.BitcoinConnection) connection) { rpc != null @@ -119,6 +120,7 @@ class UpstreamsConfigReaderSpec extends Specification { with(act.upstreams.get(0)) { id == "local" chain == "bitcoin" + options.providesBalance == true connection instanceof UpstreamsConfig.BitcoinConnection with((UpstreamsConfig.BitcoinConnection) connection) { rpc != null diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy index be866781..8b04a6bf 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy @@ -259,6 +259,7 @@ class TrackBitcoinAddressSpec extends Specification { } MultistreamHolder upstreams = new MultistreamHolderMock(Chain.BITCOIN, upstream) TrackBitcoinAddress track = new TrackBitcoinAddress(upstreams) + track.setBalanceAvailability(Chain.BITCOIN, true) when: def resp = track.subscribe(BlockchainOuterClass.BalanceRequest.newBuilder() @@ -285,6 +286,5 @@ class TrackBitcoinAddressSpec extends Specification { } .expectComplete() .verify(Duration.ofSeconds(1)) - } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilitiesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilitiesSpec.groovy new file mode 100644 index 00000000..3932c8eb --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilitiesSpec.groovy @@ -0,0 +1,32 @@ +package io.emeraldpay.dshackle.upstream.grpc + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.upstream.Capability +import spock.lang.Specification + +class RemoteCapabilitiesSpec extends Specification { + + def "parse from remote - all"() { + setup: + def remote = BlockchainOuterClass.DescribeChain.newBuilder() + .addCapabilities(BlockchainOuterClass.Capabilities.CAP_CALLS) + .addCapabilities(BlockchainOuterClass.Capabilities.CAP_NONE) + .addCapabilities(BlockchainOuterClass.Capabilities.CAP_BALANCE) + .build() + when: + def act = RemoteCapabilities.extract(remote) + then: + act == [Capability.BALANCE, Capability.RPC].toSet() + } + + def "parse from remote - only call"() { + setup: + def remote = BlockchainOuterClass.DescribeChain.newBuilder() + .addCapabilities(BlockchainOuterClass.Capabilities.CAP_CALLS) + .build() + when: + def act = RemoteCapabilities.extract(remote) + then: + act == [Capability.RPC].toSet() + } +} diff --git a/src/test/resources/upstreams-bitcoin-esplora.yaml b/src/test/resources/upstreams-bitcoin-esplora.yaml index 165158ec..ef7ceb7e 100644 --- a/src/test/resources/upstreams-bitcoin-esplora.yaml +++ b/src/test/resources/upstreams-bitcoin-esplora.yaml @@ -9,6 +9,8 @@ defaults: upstreams: - id: local chain: bitcoin + options: + balance: true connection: bitcoin: rpc: