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

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

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

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
 

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

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

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<String, Int>? = 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")
                 }

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

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

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<JsonRpcRequest, JsonRpcResponse>,
+    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) {

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<JsonRpcRequest, JsonRpcResponse>,
+    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<BlockContainer> {
+        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
+    }
+}

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<ByteArray>()
+
+    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> | <uint32 sequence number in Little Endian>
+    // i.e., it's a triple of values
+    data class Message(
+        val id: ByteArray,
+        val value: ByteArray,
+        val sequence: ByteArray,
+    )
+}

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(
                 [