Revamp dashboard and persist pool stats across restarts

Dashboard now renders 10 tiles in a 5x2 overview: hashrate, best
share, miners, network hashrate, and expected block on the top row;
difficulty, block height, block reward, total work, and the
difficulty-adjustment countdown on the bottom row. Difficulty is
rendered with T/P suffixes instead of scientific notation, the main
hashrate card shows the 1-minute value, and the block-height tile
pulses orange when the network tip advances.

Added a 24-hour hashrate area chart below the overview, sampled
once per minute. Samples are persisted to a new hashrate_samples
SQLite table and restored on startup so the chart doesn't reset
every time kamado-api is restarted.

Cumulative pool work (sum of accepted diff-1-normalized shares) is
now tracked across ckpool restarts. The aggregator integrates only
positive deltas on pool.Shares — a regression means ckpool's
counter reset to zero and the baseline is refreshed without losing
the running total. A hasPoolSharesBaseline flag prevents double-
counting on the first refresh after a kamado-api restart. The
value is persisted to a new kv table once per minute.

Next-block reward (subsidy + fees) is fetched from bitcoind
getblocktemplate at most once per minute and surfaced as a tile.

Header's block-height badge now reads prevHeight via untrack() so
the effect doesn't form a dependency cycle with its own write.
This commit is contained in:
satoshi
2026-04-22 21:07:22 +03:00
parent 8de767646d
commit e881bc930d
10 changed files with 2180 additions and 14 deletions
+170
View File
@@ -7,6 +7,7 @@ package state
import (
"context"
"log/slog"
"strconv"
"sync"
"time"
@@ -16,6 +17,12 @@ import (
"github.com/kamadopool/kamado-api/internal/zmqmon"
)
// HashratePoint is a single timestamped hashrate sample for the 24h chart.
type HashratePoint struct {
T int64 `json:"t"` // unix seconds
V float64 `json:"v"` // H/s
}
// Snapshot is the merged view served to the UI. All fields are safe to
// JSON-serialize directly.
type Snapshot struct {
@@ -34,6 +41,18 @@ type Snapshot struct {
HashrateHs1h float64 `json:"hashrate_hs_1h"`
HashrateHs24h float64 `json:"hashrate_hs_24h"`
// Best share difficulty ever seen across all workers.
BestDiff float64 `json:"best_diff"`
// Cumulative work done by the pool across its entire lifetime, in
// diff-1-normalized shares (multiply by 2^32 for total hashes).
// Survives ckpool restarts via kv-store persistence.
CumulativeShares float64 `json:"cumulative_shares"`
// Next-block reward (subsidy + fees) from bitcoind getblocktemplate,
// in BTC. Refreshed at most once per minute.
NextBlockRewardBTC float64 `json:"next_block_reward_btc"`
// Bitcoin Core fields
Chain *bitcoind.BlockchainInfo `json:"chain"`
NetworkHashrateHs float64 `json:"network_hashrate_hs"`
@@ -41,12 +60,20 @@ type Snapshot struct {
// Recent found blocks (in-memory history; persisted in Phase 2b.5).
RecentBlocks []BlockRecord `json:"recent_blocks,omitempty"`
// 24-hour hashrate history, sampled once per minute (max 1440 points).
HashrateHistory []HashratePoint `json:"hashrate_history,omitempty"`
// Health
CKPoolOK bool `json:"ckpool_ok"`
BitcoinOK bool `json:"bitcoin_ok"`
LastError string `json:"last_error,omitempty"`
}
const (
maxHistoryPoints = 1440 // 24h at 1 sample/min
kvCumulativeWork = "cumulative_shares"
)
// Aggregator refreshes a Snapshot on a ticker.
type Aggregator struct {
CK *ckpool.Client
@@ -63,6 +90,31 @@ type Aggregator struct {
snap Snapshot
blocks []BlockRecord
// 24h hashrate history ring buffer, sampled once per minute.
hrHistory []HashratePoint
lastHRSampleAt time.Time
// Cumulative pool work in diff-1-normalized shares, surviving
// ckpool restarts. Maintained by re-reading pool.Shares each
// refresh and integrating only the positive delta — a decrease
// means ckpool restarted and its counter reset to 0, so we reset
// our delta baseline without losing the accumulated total.
//
// hasPoolSharesBaseline guards against double-counting on api
// restart: cumulativeShares is loaded from disk, and the first
// post-load observation of pool.Shares only establishes the
// baseline — the current ckpool counter was already accounted for
// before we crashed.
cumulativeShares float64
lastPoolShares float64
hasPoolSharesBaseline bool
lastCumulativeSave time.Time
// Next-block reward cache; refreshed on its own cadence so we don't
// hammer bitcoind with getblocktemplate on every poll tick.
nextBlockReward float64
lastTemplateFetch time.Time
// ckFailStreak counts consecutive refreshes where CKPool returned an
// error. The first failure after a streak of successes is logged at
// DEBUG (likely a transient warm-up or lock hiccup); repeated failures
@@ -86,6 +138,7 @@ func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog
// 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.loadPersistedState()
a.refresh(ctx)
t := time.NewTicker(a.Interval)
defer t.Stop()
@@ -167,6 +220,88 @@ func (a *Aggregator) refresh(ctx context.Context) {
}
}
// Compute pool-wide best-ever share diff from workers.
for _, w := range next.Workers {
d := w.BestEver
if d == 0 {
d = w.BestDiff
}
if d > next.BestDiff {
next.BestDiff = d
}
}
a.mu.Lock()
now := time.Now()
// --- cumulative work tracking ---
// Integrate only positive deltas on pool.Shares. A decrease means
// ckpool restarted (or its state reset); reset the baseline without
// touching the accumulated total. The first observation after load
// only establishes the baseline — those shares were already
// accumulated before the previous shutdown.
if next.Pool != nil {
cur := float64(next.Pool.Shares)
if a.hasPoolSharesBaseline && cur >= a.lastPoolShares {
a.cumulativeShares += cur - a.lastPoolShares
}
a.lastPoolShares = cur
a.hasPoolSharesBaseline = true
}
next.CumulativeShares = a.cumulativeShares
// Persist cumulative_shares at most once per minute.
if a.Store != nil && now.Sub(a.lastCumulativeSave) >= time.Minute {
val := strconv.FormatFloat(a.cumulativeShares, 'f', -1, 64)
if err := a.Store.SetKV(kvCumulativeWork, val); err != nil {
a.Log.Warn("cumulative_shares persist failed", "err", err)
}
a.lastCumulativeSave = now
}
// --- hashrate history sample (once per minute) ---
if now.Sub(a.lastHRSampleAt) >= time.Minute {
p := HashratePoint{T: now.Unix(), V: next.HashrateHs}
a.hrHistory = append(a.hrHistory, p)
if len(a.hrHistory) > maxHistoryPoints {
a.hrHistory = a.hrHistory[len(a.hrHistory)-maxHistoryPoints:]
}
a.lastHRSampleAt = now
if a.Store != nil {
if err := a.Store.InsertHashrateSample(p.T, p.V); err != nil {
a.Log.Warn("hashrate persist failed", "err", err)
}
// Keep the persisted window bounded to 24h + a small slack.
cutoff := now.Add(-25 * time.Hour).Unix()
_ = a.Store.PruneHashrateBefore(cutoff)
}
}
if len(a.hrHistory) > 0 {
next.HashrateHistory = make([]HashratePoint, len(a.hrHistory))
copy(next.HashrateHistory, a.hrHistory)
}
// --- next-block reward (at most once per minute) ---
next.NextBlockRewardBTC = a.nextBlockReward
needTemplate := now.Sub(a.lastTemplateFetch) >= time.Minute
a.mu.Unlock()
if needTemplate && a.RPC != nil && next.BitcoinOK {
tplCtx, cancelTpl := context.WithTimeout(ctx, 5*time.Second)
tpl, err := a.RPC.GetBlockTemplate(tplCtx)
cancelTpl()
if err == nil && tpl != nil {
reward := float64(tpl.CoinbaseValue) / 1e8
a.mu.Lock()
a.nextBlockReward = reward
a.lastTemplateFetch = now
next.NextBlockRewardBTC = reward
a.mu.Unlock()
} else if err != nil {
a.Log.Debug("getblocktemplate failed", "err", err)
}
}
a.mu.Lock()
// Attach current block history so /api/snapshot and WebSocket
// pushes carry the same view.
@@ -183,3 +318,38 @@ func (a *Aggregator) refresh(ctx context.Context) {
cb(pushed)
}
}
// loadPersistedState restores cumulative work and hashrate history from
// the store so they survive process restarts. Safe to call with a nil
// Store — becomes a no-op.
func (a *Aggregator) loadPersistedState() {
if a.Store == nil {
return
}
if v, err := a.Store.GetKV(kvCumulativeWork); err != nil {
a.Log.Warn("cumulative_shares load failed", "err", err)
} else if v != "" {
if f, perr := strconv.ParseFloat(v, 64); perr == nil {
a.mu.Lock()
a.cumulativeShares = f
a.mu.Unlock()
a.Log.Info("cumulative_shares loaded", "value", f)
}
}
cutoff := time.Now().Add(-24 * time.Hour).Unix()
if samples, err := a.Store.HashrateSince(cutoff); err != nil {
a.Log.Warn("hashrate history load failed", "err", err)
} else if len(samples) > 0 {
hist := make([]HashratePoint, 0, len(samples))
for _, s := range samples {
hist = append(hist, HashratePoint{T: s.T, V: s.V})
}
a.mu.Lock()
a.hrHistory = hist
a.lastHRSampleAt = time.Unix(hist[len(hist)-1].T, 0)
a.mu.Unlock()
a.Log.Info("hashrate history loaded", "count", len(hist))
}
}