From ede552be591bb75edbc870325562e02758c726a7 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sat, 25 Jun 2022 21:26:10 -0400 Subject: [PATCH] problem: Bitcoin head check may lag because it checks it only once in 15 seconds solution: use ZeroMQ connection to listen for new blocks --- README.adoc | 6 - build.gradle | 1 + docs/reference-configuration.adoc | 44 ++++++- gradle/libs.versions.toml | 2 + .../dshackle/config/UpstreamsConfig.kt | 6 + .../dshackle/config/UpstreamsConfigReader.kt | 19 +++ .../dshackle/startup/ConfiguredUpstreams.kt | 18 ++- .../upstream/CurrentMultistreamHolder.kt | 12 ++ .../upstream/bitcoin/BitcoinRpcUpstream.kt | 2 +- .../upstream/bitcoin/BitcoinZMQHead.kt | 67 ++++++++++ .../dshackle/upstream/bitcoin/ZMQServer.kt | 116 ++++++++++++++++++ .../startup/ConfiguredUpstreamsSpec.groovy | 4 +- 12 files changed, 283 insertions(+), 14 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt diff --git a/README.adoc b/README.adoc index dad56ca9..f054235d 100644 --- a/README.adoc +++ b/README.adoc @@ -44,14 +44,8 @@ Blockchains support: image::dshackle-intro.png[alt="",width=80%,align="center"] -== Roadmap - WARNING: The project is still under development, please use with caution. -- [ ] Subscription to bitcoind notification over gRPC (instead of ZeroMQ) -- [ ] Lightweight sidecar node connector -- [ ] Configurable upstream roles - == Quick Start === Configuration diff --git a/build.gradle b/build.gradle index 8e213420..5ec8d72f 100644 --- a/build.gradle +++ b/build.gradle @@ -56,6 +56,7 @@ dependencies { implementation libs.bundles.grpc implementation libs.bundles.netty + implementation libs.zeromq implementation(libs.bundles.spring.framework) { exclude module: 'spring-boot-starter-logging' } diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index 4410c48a..c6e5e773 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -142,9 +142,12 @@ cluster: basic-auth: username: bitcoin password: e984af45bb888428207c290 - # uses Esplora index to fetch balances and utxo for an address + # use Esplora index to fetch balances and utxo for an address esplora: url: "http://localhost:3001" + # connect via ZeroMQ to get notifications about new blocks + zeromq: + address: "http://localhost:5555" - id: remote connection: grpc: @@ -720,7 +723,9 @@ See link:09-quorum-and-selectors.adoc[Quorum and Selectors] |=== -.Connection Config +==== Ethereum Connection Options + +.Connection Config for Ethereum Upstream [cols="2a,5"] |=== | Option | Description @@ -764,6 +769,41 @@ Default is 15Mb |=== +==== Bitcoin Connection Options + +.Connection Config for Bitcoin Upstream +[cols="2a,5"] +|=== +| Option | Description + +| `rpc.url` +a| HTTP URL to connect to. This is required for a connection. + +URL can be configured with Environment Variable placeholders `${ENV_VAR_NAME}`. + +Example: `http://${NODE_HOST}:${NODE_PORT}` + +| `rpc.basic-auth` + `rpc.basic-auth.username`, `rpc.basic-auth.password` +a| HTTP Basic Auth configuration, which is required by the Bitcoind server. + +Values can also reference env variables, for example: +[source,yaml] +---- +rpc: + url: "http://127.0.0.1:8332" + basic-auth: + username: "${NODE_USERNAME}" + password: "${NODE_PASSWORD}" +---- + +| `zeromq.address` +a| Set up an additional connection via ZeroMQ protocol to subscribe to the new blocks. +The node must be launched with the same address specified as `-zmqpubhashblock="tcp://${HOST}:${POST}"` or in `bitcoin.conf` +[source,yaml] +---- +zeromq: + address: "127.0.0.1:5555" +---- + +|=== + [#upstream-dshackle] === Dshackle Upstream diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index f77cf411..5924c7f7 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -86,6 +86,8 @@ netty-buffer = { module = "io.netty:netty-buffer", version.ref = "netty" } netty-tcnative-core = { module = "io.netty:netty-tcnative", version.ref = "netty-tcnative" } netty-tcnative-boringssl = { module = "io.netty:netty-tcnative-boringssl-static", version.ref = "netty-tcnative" } +zeromq = "org.zeromq:jeromq:0.5.2" + objgenesis = "org.objenesis:objenesis:3.1" reactor-core = { module = "io.projectreactor:reactor-core", version.ref = "reactor" } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 552a6ddf..d681a5bb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -111,8 +111,14 @@ open class UpstreamsConfig { class BitcoinConnection : RpcConnection() { var esplora: HttpEndpoint? = null + var zeroMq: BitcoinZeroMq? = null } + data class BitcoinZeroMq( + val host: String = "127.0.0.1", + val port: Int + ) + class HttpEndpoint(val url: URI) { var basicAuth: AuthConfig.ClientBasicAuth? = null var tls: AuthConfig.ClientTlsAuth? = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index a0fd5e8e..a118d956 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -153,6 +153,25 @@ class UpstreamsConfigReader( connection.esplora = http } } + getMapping(connConfigNode, "zeromq")?.let { node -> + getValueAsString(node, "address")?.let { address -> + val zmqConfig: Pair? = try { + if (address.contains(":")) { + address.split(":").let { + Pair(it[0], it[1].toInt()) + } + } else { + Pair("127.0.0.1", address.toInt()) + } + } catch (t: Throwable) { + log.warn("Invalid config for ZeroMQ: $address. Expected to be in format HOST:PORT") + null + } + zmqConfig?.let { + connection.zeroMq = UpstreamsConfig.BitcoinZeroMq(it.first, it.second) + } + } + } } else { log.error("Upstream at #0 has invalid configuration") } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index c02f2340..48e4f189 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -18,12 +18,17 @@ package io.emeraldpay.dshackle.startup import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.MergedHead +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead import io.emeraldpay.dshackle.upstream.bitcoin.EsploraClient +import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock +import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream @@ -52,7 +57,6 @@ open class ConfiguredUpstreams( @Autowired private val currentUpstreams: CurrentMultistreamHolder, @Autowired private val fileResolver: FileResolver, @Autowired private val config: UpstreamsConfig, - @Autowired private val cachesFactory: CachesFactory ) { private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) @@ -159,11 +163,19 @@ open class ConfiguredUpstreams( EsploraClient(endpoint.url, endpoint.basicAuth, tls) } + val extractBlock = ExtractBlock() + val rpcHead = BitcoinRpcHead(directApi, extractBlock) + val head: Head = conn.zeroMq?.let { zeroMq -> + val server = ZMQServer(zeroMq.host, zeroMq.port, "hashblock") + val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock) + MergedHead(listOf(rpcHead, zeroMqHead)) + } ?: rpcHead + val methods = buildMethods(config, chain) val upstream = BitcoinRpcUpstream( config.id ?: "bitcoin-${seq.getAndIncrement()}", - chain, directApi, + chain, directApi, head, options, config.role, QuorumForLabels.QuorumItem(1, config.labels), methods, esplora diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index f52da7ce..9a99d6de 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt @@ -37,6 +37,7 @@ import java.util.Collections import java.util.concurrent.Callable import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.locks.ReentrantLock +import javax.annotation.PreDestroy import kotlin.concurrent.withLock @Repository @@ -145,4 +146,15 @@ open class CurrentMultistreamHolder( override fun isAvailable(chain: Chain): Boolean { return chainMapping.containsKey(chain) && callTargets.containsKey(chain) } + + @PreDestroy + fun shutdown() { + log.info("Closing upstream connections...") + updateLock.withLock { + chainMapping.values.forEach { + it.stop() + } + chainMapping.clear() + } + } } 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 18afd4f1..e5ab2d29 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt @@ -34,6 +34,7 @@ open class BitcoinRpcUpstream( id: String, chain: Chain, private val directApi: Reader, + private val head: Head, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, node: QuorumForLabels.QuorumItem, @@ -45,7 +46,6 @@ open class BitcoinRpcUpstream( private val log = LoggerFactory.getLogger(BitcoinRpcUpstream::class.java) } - private val head: Head = createHead() private var validatorSubscription: Disposable? = null private val capabilities = if (options.providesBalance == true) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt new file mode 100644 index 00000000..fa0d206d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt @@ -0,0 +1,67 @@ +package io.emeraldpay.dshackle.upstream.bitcoin + +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.AbstractHead +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import org.apache.commons.codec.binary.Hex +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.util.retry.Retry +import java.time.Duration + +class BitcoinZMQHead( + private val server: ZMQServer, + private val api: Reader, + private val extractBlock: ExtractBlock, +) : Head, AbstractHead(), Lifecycle { + + companion object { + private val log = LoggerFactory.getLogger(BitcoinZMQHead::class.java) + } + + private var refreshSubscription: Disposable? = null + + fun connect(): Flux { + return Flux.from(server.sink.asFlux()) + .onBackpressureLatest() + .map { + Hex.encodeHexString(it) + } + .flatMap { hash -> + api.read(JsonRpcRequest("getblock", listOf(hash))) + .switchIfEmpty(Mono.error(IllegalStateException("Block $hash is not available on upstream"))) + .retryWhen(Retry.backoff(5, Duration.ofMillis(100))) + .switchIfEmpty(Mono.fromCallable { log.warn("Block $hash is not available on upstream") }.then(Mono.empty())) + .flatMap(JsonRpcResponse::requireResult) + .map(extractBlock::extract) + .timeout(Defaults.timeout, Mono.error(Exception("Block data is not received"))) + } + .onErrorResume { t -> + log.warn("Failed to get a block from upstream with error: ${t.message}") + connect() + } + } + + override fun start() { + server.start() + refreshSubscription = super.follow(connect()) + } + + override fun stop() { + server.stop() + val copy = refreshSubscription + refreshSubscription = null + copy?.dispose() + } + + override fun isRunning(): Boolean { + return server.isRunning || refreshSubscription != null + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt new file mode 100644 index 00000000..1358f568 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt @@ -0,0 +1,116 @@ +package io.emeraldpay.dshackle.upstream.bitcoin + +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import org.zeromq.SocketType +import org.zeromq.ZContext +import org.zeromq.ZMQ +import reactor.core.publisher.Sinks +import java.util.concurrent.Executors +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +class ZMQServer( + val host: String, + val port: Int, + val topic: String, +) : Lifecycle { + + companion object { + private val log = LoggerFactory.getLogger(ZMQServer::class.java) + private val RECEIVE_TIME_MS = 100 + } + + private val topicId = topic.encodeToByteArray() + + private val runningLock = ReentrantLock() + private var running = false + private val sinkPublisher = Executors.newSingleThreadExecutor() + val sink = Sinks.many() + .multicast() + .directBestEffort() + + fun startInternal() = Runnable { + log.info("Connecting to ZMQ at $host:$port") + val context = ZContext() + val socket: ZMQ.Socket = context.createSocket(SocketType.SUB) + socket.connect("tcp://$host:$port") + socket.subscribe(topic) + socket.receiveTimeOut = RECEIVE_TIME_MS + + while (running && !Thread.currentThread().isInterrupted) { + val msg = readMessage(socket) + if (msg != null) { + // this should not happen, but check just in case + if (!topicId.contentEquals(msg.id)) { + continue + } + sinkPublisher.execute { + val sent = sink.tryEmitNext(msg.value) + if (sent.isFailure && sent != Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) { + log.warn("Failed to notify with $sent") + } + } + } + } + socket.close() + log.debug("Stopped ZMQ connection to $host:$port") + } + + fun readMessage(socket: ZMQ.Socket): Message? { + val id = readOnce(socket) + val value = readOnce(socket) + val seq = readOnce(socket) + if (id != null && value != null && seq != null) { + return Message(id, value, seq) + } + return null + } + + fun readOnce(socket: ZMQ.Socket): ByteArray? { + while (running) { + // blocks until a message is received + // but usually returns null if nothing received, so have to repeat a request + val data = socket.recv(0) + if (data != null) { + return data + } + } + return null + } + + override fun start() { + runningLock.withLock { + if (running) { + return + } + running = true + } + Thread(startInternal()).start() + } + + override fun stop() { + runningLock.withLock { + if (!running) { + return + } + running = false + } + log.debug("Stopping ZMQ listener at $host:$port") + // give some time to the internal thread to receive a message and quit + Thread.sleep(RECEIVE_TIME_MS.toLong()) + } + + override fun isRunning(): Boolean { + return running + } + + // Bitcoind produces messages as something like: + // | hashblock | <32-byte block hash in Little Endian> | + // i.e., it's a triple of values + data class Message( + val id: ByteArray, + val value: ByteArray, + val sequence: ByteArray, + ) +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index 6694b2e8..106541c6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -18,7 +18,7 @@ class ConfiguredUpstreamsSpec extends Specification { _ * getDefaultMethods(Chain.ETHEREUM) >> new DefaultEthereumMethods(Chain.ETHEREUM) } def configurer = new ConfiguredUpstreams( - currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig), Stub(CachesFactory) + currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig) ) def methods = new UpstreamsConfig.Methods( [ @@ -42,7 +42,7 @@ class ConfiguredUpstreamsSpec extends Specification { _ * getDefaultMethods(Chain.ETHEREUM) >> new DefaultEthereumMethods(Chain.ETHEREUM) } def configurer = new ConfiguredUpstreams( - currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig), Stub(CachesFactory) + currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig) ) def methods = new UpstreamsConfig.Methods( [