diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go index c55d2c4..fef9166 100644 --- a/api/internal/state/aggregator.go +++ b/api/internal/state/aggregator.go @@ -270,6 +270,13 @@ type Aggregator struct { latencyLastMs int64 // most recent observation staleWorkHashes float64 // cumulative wasted hashes = sum(latency_s * hashrate_at_event) + // notifierDone is signalled by IngestLatencyEvents when a latency + // event arrives, meaning ckpool's notifier has finished its + // getblocktemplate + mining.notify cycle. The Run loop uses this to + // defer dashboard RPC calls until after the notifier is done, + // avoiding resource contention on bitcoind. + notifierDone chan struct{} + // High-water mark for the best share difficulty ever seen. // Persisted to kv so it survives restarts and transient ckpool gaps. bestDiff float64 @@ -296,13 +303,14 @@ type Aggregator struct { func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator { return &Aggregator{ - CK: ck, - RPC: rpc, - Interval: interval, - Log: log, + CK: ck, + RPC: rpc, + Interval: interval, + Log: log, sessionRejectReasons: make(map[string]int64), alltimeRejectReasons: make(map[string]int64), - ready: make(chan struct{}), + ready: make(chan struct{}), + notifierDone: make(chan struct{}, 1), } } @@ -345,7 +353,21 @@ func (a *Aggregator) Run(ctx context.Context, tipEvents <-chan zmqmon.TipEvent) a.mu.Lock() a.lastZMQEventTime = ev.SeenAt a.mu.Unlock() - a.Log.Debug("zmq tip, refreshing", "hash", ev.Hash) + // Wait for ckpool's notifier to finish (getblocktemplate + + // mining.notify) before firing our dashboard RPCs, so we + // don't compete for bitcoind RPC resources during the + // latency-critical window. The latency log line signals + // completion. Timeout after 5s in case the log line is + // missed or ckpool is slow. + a.Log.Debug("zmq tip, waiting for notifier", "hash", ev.Hash) + select { + case <-a.notifierDone: + a.Log.Debug("notifier done, refreshing dashboard") + case <-time.After(5 * time.Second): + a.Log.Debug("notifier wait timed out, refreshing anyway") + case <-ctx.Done(): + return + } a.refresh(ctx) } } diff --git a/api/internal/state/blocks.go b/api/internal/state/blocks.go index 72375a2..1ea5d62 100644 --- a/api/internal/state/blocks.go +++ b/api/internal/state/blocks.go @@ -103,6 +103,12 @@ func (a *Aggregator) IngestLatencyEvents(ctx context.Context, events <-chan logm a.staleWorkHashes += (float64(ev.LatencyMs) / 1000.0) * hs5m count, sum, last, stale := a.latencyCount, a.latencySumMs, a.latencyLastMs, a.staleWorkHashes a.mu.Unlock() + // Signal the Run loop that ckpool's notifier is done so + // dashboard RPC calls can proceed without competing. + select { + case a.notifierDone <- struct{}{}: + default: + } a.Log.Info("logmon: block update latency", "ms", ev.LatencyMs, "count", count) // Blocks are infrequent (~10min); persist every event. if a.Store != nil {