From b786f489a11550dfc76c2de392c0ca067d516bda Mon Sep 17 00:00:00 2001 From: satoshi Date: Tue, 14 Apr 2026 11:14:52 +0300 Subject: [PATCH] Add ZMQ hashblock subscriber for sub-second chain refresh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- api/cmd/kamado-api/main.go | 10 ++- api/go.mod | 8 +- api/go.sum | 8 ++ api/internal/state/aggregator.go | 13 +++- api/internal/zmqmon/zmqmon.go | 122 +++++++++++++++++++++++++++++++ 5 files changed, 156 insertions(+), 5 deletions(-) create mode 100644 api/internal/zmqmon/zmqmon.go diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go index 3800c89..49d8f83 100644 --- a/api/cmd/kamado-api/main.go +++ b/api/cmd/kamado-api/main.go @@ -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) diff --git a/api/go.mod b/api/go.mod index e66a5e8..1c66377 100644 --- a/api/go.mod +++ b/api/go.mod @@ -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 diff --git a/api/go.sum b/api/go.sum index 5424fe4..007321b 100644 --- a/api/go.sum +++ b/api/go.sum @@ -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= diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go index 81f3f43..6bcfb58 100644 --- a/api/internal/state/aggregator.go +++ b/api/internal/state/aggregator.go @@ -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) } } } diff --git a/api/internal/zmqmon/zmqmon.go b/api/internal/zmqmon/zmqmon.go new file mode 100644 index 0000000..41f7507 --- /dev/null +++ b/api/internal/zmqmon/zmqmon.go @@ -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) + } + } +}