satoshi
b786f489a1
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.
123 lines
3.2 KiB
Go
123 lines
3.2 KiB
Go
// 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)
|
|
}
|
|
}
|
|
}
|