StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: delegate bitcoin balance request to another upstream, if data is unavailable locally

Browse Source
This commit is contained in:
Igor Artamonov
2020-09-22 23:51:07 -04:00
parent 19717d622b
commit d537ff0461
20 changed files with 276 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/kotlin/io/emeraldpay/dshackle/SilentException.kt
View File

@@ -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")
}

2
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -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
}

3
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -261,6 +261,9 @@ class UpstreamsConfigReader(
getValueAsBool(values, "disable-validation")?.let {
options.disableValidation = it
}
getValueAsBool(values, "balance")?.let {
options.providesBalance = it
}
return options
}

11
src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
View File

@@ -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<Capability> = 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())
}
}

120
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
View File

@@ -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<Chain, Boolean> = 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<String> {
if (!request.hasAddress()) {
return Flux.empty()
@@ -117,38 +161,74 @@ class TrackBitcoinAddress(
}
}
fun getBalanceGrpc(api: BitcoinMultistream): Mono<ReactorBlockchainGrpc.ReactorBlockchainStub> {
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<BlockchainOuterClass.AddressBalance> {
return getBalanceGrpc(api).flatMapMany { remote ->
remote.getBalance(request)
}
}
fun subscribeRemoteBalance(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
return getBalanceGrpc(api).flatMapMany { remote ->
remote.subscribeBalance(request)
}
}
override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
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<BlockchainOuterClass.AddressBalance> {
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<String, BigInteger>()
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<String, BigInteger>()
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 {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -127,7 +127,7 @@ class CurrentMultistreamHolder(
}
override fun observeChains(): Flux<Chain> {
return Flux.merge(
return Flux.concat(
Flux.fromIterable(getAvailable()),
Flux.from(chainsBus)
)

19
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -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<Capability> = 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<Capability> {
return this.capabilities
}
override fun isGrpc(): Boolean {
return false
}
fun printStatus() {
var height: Long? = null
try {

25
src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
View File

@@ -149,10 +149,18 @@ class Selector {
}
}
class LabelMatcher(val name: String, val values: Collection<String>): 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<String>) : 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()
}
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
View File

@@ -37,6 +37,8 @@ interface Upstream {
fun getLabels(): Collection<UpstreamsConfig.Labels>
fun getMethods(): CallMethods
fun getId(): String
fun getCapabilities(): Set<Capability>
fun isGrpc(): Boolean
fun <T : Upstream> cast(selfType: Class<T>): T
}

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
View File

@@ -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<Capability> {
return capabilities;
}
override fun isGrpc(): Boolean {
return false
}
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
View File

@@ -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<Capability> {
return capabilities
}
override fun isGrpc(): Boolean {
return false
}
@Suppress("unchecked")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {

17
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -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<Capability> = emptySet()
override fun getHead(): Head {
return grpcHead
@@ -105,6 +103,14 @@ class BitcoinGrpcUpstream(
return upstreamStatus.getLabels()
}
override fun getCapabilities(): Set<Capability> {
return capabilities
}
override fun isGrpc(): Boolean {
return true
}
override fun <T : Upstream> cast(selfType: Class<T>): 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) }
}

9
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
View File

@@ -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<Capability> = emptySet()
private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = 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<Capability> {
return capabilities
}
override fun isGrpc(): Boolean {
return true
}
}

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt Normal file
View File

@@ -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<Capability> {
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()
}
}
}

2
src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
View File

@@ -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

2
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy
View File

@@ -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))
}
}

32
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilitiesSpec.groovy Normal file
View File

@@ -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()
}
}

2
src/test/resources/upstreams-bitcoin-esplora.yaml
View File

@@ -9,6 +9,8 @@ defaults:
upstreams:
- id: local
chain: bitcoin
options:
balance: true
connection:
bitcoin:
rpc: