satoshi/KamadoPool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add ZMQ hashblock subscriber for sub-second chain refresh

Browse Source 
New internal/zmqmon package subscribes to bitcoind's hashblock
ZMQ topic and emits TipEvents. Uses github.com/go-zeromq/zmq4
(pure Go, builds with CGO_ENABLED=0). Exponential backoff on
connection failure so bitcoind restarts don't kill the
subscriber permanently; event channel drops rather than blocks
if the consumer is slow (signals are advisory, not logs).

Aggregator.Run now takes a <-chan TipEvent; when a tip arrives
it fires an immediate refresh() outside the normal ticker
cadence. With a 5s poll interval and 0.5-1s ZMQ latency from
bitcoind, dashboards now reflect new tips roughly 4x faster.

Endpoint comes from BITCOIN_ZMQ_BLOCK — empty disables ZMQ
entirely and the aggregator just runs on the ticker alone.
This commit is contained in:
satoshi
2026-04-14 11:14:52 +03:00
parent 01829746ac
commit b786f489a1
5 changed files with 156 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/cmd/kamado-api/main.go
+8 -2
View File
@@ -22,6 +22,7 @@ import (
"github.com/kamadopool/kamado-api/internal/logmon"
"github.com/kamadopool/kamado-api/internal/state"
"github.com/kamadopool/kamado-api/internal/store"
"github.com/kamadopool/kamado-api/internal/zmqmon"
)
func main() {
@@ -69,9 +70,14 @@ func main() {
// real-time updates without polling.
agg.OnRefresh = api.Hub.Broadcast
go agg.Run(ctx)
// Optional: bitcoind hashblock ZMQ subscription for sub-second
// chain-tip refreshes. Empty endpoint disables it.
zmq := zmqmon.New(cfg.BitcoinZMQBlock, log)
go zmq.Run(ctx)
// Tail the ckpool log for block-solve events.
go agg.Run(ctx, zmq.Events)
// Tail the ckpool log for block-solve events (our own solves).
tailer := logmon.New(cfg.CKPoolLogFile, log)
go tailer.Run(ctx)
go agg.IngestBlockEvents(ctx, tailer.Events)
api/go.mod
+7 -1
View File
@@ -2,15 +2,21 @@ module github.com/kamadopool/kamado-api
go 1.22
require modernc.org/sqlite v1.34.5
require (
github.com/go-zeromq/zmq4 v0.17.0
modernc.org/sqlite v1.34.5
)
require (
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/go-zeromq/goczmq/v4 v4.2.2 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/ncruces/go-strftime v0.1.9 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
golang.org/x/sync v0.7.0 // indirect
golang.org/x/sys v0.22.0 // indirect
golang.org/x/text v0.15.0 // indirect
modernc.org/libc v1.55.3 // indirect
modernc.org/mathutil v1.6.0 // indirect
modernc.org/memory v1.8.0 // indirect
api/go.sum
+8
View File
@@ -1,5 +1,9 @@
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/go-zeromq/goczmq/v4 v4.2.2 h1:HAJN+i+3NW55ijMJJhk7oWxHKXgAuSBkoFfvr8bYj4U=
github.com/go-zeromq/goczmq/v4 v4.2.2/go.mod h1:Sm/lxrfxP/Oxqs0tnHD6WAhwkWrx+S+1MRrKzcxoaYE=
github.com/go-zeromq/zmq4 v0.17.0 h1:r12/XdqPeRbuaF4C3QZJeWCt7a5vpJbslDH1rTXF+Kc=
github.com/go-zeromq/zmq4 v0.17.0/go.mod h1:EQxjJD92qKnrsVMzAnx62giD6uJIPi1dMGZ781iCDtY=
github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo=
github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -12,9 +16,13 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
golang.org/x/mod v0.16.0 h1:QX4fJ0Rr5cPQCF7O9lh9Se4pmwfwskqZfq5moyldzic=
golang.org/x/mod v0.16.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.22.0 h1:RI27ohtqKCnwULzJLqkv897zojh5/DwS/ENaMzUOaWI=
golang.org/x/sys v0.22.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw=
golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc=
modernc.org/cc/v4 v4.21.4 h1:3Be/Rdo1fpr8GrQ7IVw9OHtplU4gWbb+wNgeoBMmGLQ=
api/internal/state/aggregator.go
+11 -2
View File
@@ -13,6 +13,7 @@ import (
"github.com/kamadopool/kamado-api/internal/bitcoind"
"github.com/kamadopool/kamado-api/internal/ckpool"
"github.com/kamadopool/kamado-api/internal/store"
"github.com/kamadopool/kamado-api/internal/zmqmon"
)
// Snapshot is the merged view served to the UI. All fields are safe to
@@ -81,8 +82,9 @@ func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog
// Run blocks until ctx is cancelled, refreshing the snapshot every Interval.
// It runs one immediate refresh at startup so readers don't see an empty
// snapshot after ctx launches the goroutine. Persisted block history is
// loaded from the store before the first refresh.
func (a *Aggregator) Run(ctx context.Context) {
// loaded from the store before the first refresh. If tipEvents is non-nil,
// each received tip triggers an immediate refresh outside the poll cadence.
func (a *Aggregator) Run(ctx context.Context, tipEvents <-chan zmqmon.TipEvent) {
a.loadPersistedBlocks()
a.refresh(ctx)
t := time.NewTicker(a.Interval)
@@ -93,6 +95,13 @@ func (a *Aggregator) Run(ctx context.Context) {
return
case <-t.C:
a.refresh(ctx)
case ev, ok := <-tipEvents:
if !ok {
tipEvents = nil
continue
}
a.Log.Debug("zmq tip, refreshing", "hash", ev.Hash)
a.refresh(ctx)
}
}
}
api/internal/zmqmon/zmqmon.go
+122
View File
@@ -0,0 +1,122 @@
// Package zmqmon subscribes to bitcoind's ZMQ `hashblock` topic and
// emits a lightweight event each time a new chain tip appears.
//
// This is strictly an "immediate refresh trigger" — it does NOT replace
// the ckpool log tailer, which still handles our own solved-block
// detection. ZMQ just shortens the latency between bitcoind seeing a
// new tip and the Kamado dashboard reflecting it (otherwise we'd wait
// up to PollInterval seconds for the next state refresh).
package zmqmon
import (
"context"
"encoding/hex"
"fmt"
"log/slog"
"time"
"github.com/go-zeromq/zmq4"
)
// TipEvent is emitted when bitcoind publishes a new block hash.
type TipEvent struct {
Hash string // lowercase hex, no 0x prefix
SeenAt time.Time
}
// Monitor subscribes to the hashblock topic at Endpoint (e.g.
// "tcp://bitcoind:28332") and publishes TipEvents to Events. Run blocks
// until ctx is cancelled.
type Monitor struct {
Endpoint string
Log *slog.Logger
Events chan TipEvent
}
// New returns a Monitor with a buffered event channel. Endpoint may be
// empty, in which case Run exits immediately as a no-op — useful for
// deployments where ZMQ isn't configured on bitcoind.
func New(endpoint string, log *slog.Logger) *Monitor {
return &Monitor{
Endpoint: endpoint,
Log: log,
Events: make(chan TipEvent, 16),
}
}
// Run connects, subscribes to "hashblock", and relays tip hashes until
// ctx is cancelled. On connection failure it backs off and retries so
// bitcoind restarts don't kill the subscriber permanently.
func (m *Monitor) Run(ctx context.Context) {
defer close(m.Events)
if m.Endpoint == "" {
m.Log.Info("zmqmon disabled (no endpoint)")
return
}
backoff := time.Second
const maxBackoff = 30 * time.Second
for {
if err := m.runOnce(ctx); err != nil && ctx.Err() == nil {
m.Log.Warn("zmqmon reconnecting", "err", err, "backoff", backoff)
select {
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff < maxBackoff {
backoff *= 2
if backoff > maxBackoff {
backoff = maxBackoff
}
}
continue
}
// ctx cancelled
return
}
}
func (m *Monitor) runOnce(ctx context.Context) error {
sub := zmq4.NewSub(ctx)
defer sub.Close()
if err := sub.Dial(m.Endpoint); err != nil {
return fmt.Errorf("dial %s: %w", m.Endpoint, err)
}
if err := sub.SetOption(zmq4.OptionSubscribe, "hashblock"); err != nil {
return fmt.Errorf("subscribe: %w", err)
}
m.Log.Info("zmqmon connected", "endpoint", m.Endpoint)
for {
// Recv blocks until a message arrives or the underlying context
// is cancelled.
msg, err := sub.Recv()
if err != nil {
if ctx.Err() != nil {
return nil
}
return fmt.Errorf("recv: %w", err)
}
// bitcoind publishes 3 frames: topic, body, sequence.
if len(msg.Frames) < 2 {
continue
}
topic := string(msg.Frames[0])
if topic != "hashblock" {
continue
}
hash := hex.EncodeToString(msg.Frames[1])
select {
case m.Events <- TipEvent{Hash: hash, SeenAt: time.Now()}:
case <-ctx.Done():
return nil
default:
// Dropping is fine: the consumer only uses this as a
// "refresh now" signal, not as an event log.
m.Log.Debug("zmqmon event dropped (consumer slow)", "hash", hash)
}
}
}