diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go index fc56a6a..725d41a 100644 --- a/api/cmd/kamado-api/main.go +++ b/api/cmd/kamado-api/main.go @@ -128,6 +128,7 @@ func main() { go tailer.Run(ctx) go agg.IngestBlockEvents(ctx, tailer.Events) go agg.IngestAttemptEvents(ctx, tailer.Attempts) + go agg.IngestLatencyEvents(ctx, tailer.Latencies) srv := &http.Server{ Addr: cfg.ListenAddr, diff --git a/api/internal/ckpool/types.go b/api/internal/ckpool/types.go index 3d36e36..93787b5 100644 --- a/api/internal/ckpool/types.go +++ b/api/internal/ckpool/types.go @@ -26,6 +26,7 @@ type PoolStats struct { SPS60 float64 `json:"sps60"` Accepted int64 `json:"accepted"` Rejected int64 `json:"rejected"` + RejectCount int64 `json:"rejectcount"` DSPS1 float64 `json:"dsps1"` DSPS5 float64 `json:"dsps5"` DSPS15 float64 `json:"dsps15"` diff --git a/api/internal/logmon/tailer.go b/api/internal/logmon/tailer.go index a6f1309..ad56ca7 100644 --- a/api/internal/logmon/tailer.go +++ b/api/internal/logmon/tailer.go @@ -49,14 +49,23 @@ type AttemptEvent struct { RawLine string } +// LatencyEvent is emitted when ckpool logs the "Block update latency" +// line from our patch 0005 — the milliseconds between ZMQ trigger and +// mining.notify broadcast completion. +type LatencyEvent struct { + SeenAt time.Time + LatencyMs int64 +} + // Tailer follows a log file, surviving rotation/truncation, and emits // parsed events. Create with New, then Run in a goroutine. type Tailer struct { - Path string - Events chan BlockEvent - Attempts chan AttemptEvent - Log *slog.Logger - PollWait time.Duration // how long to sleep between EOF polls + Path string + Events chan BlockEvent + Attempts chan AttemptEvent + Latencies chan LatencyEvent + Log *slog.Logger + PollWait time.Duration // how long to sleep between EOF polls // LoadCursor / SaveCursor, if both non-nil, persist the read // position across process restarts. LoadCursor returns (inode, @@ -77,11 +86,12 @@ type Tailer struct { func New(path string, log *slog.Logger) *Tailer { return &Tailer{ - Path: path, - Events: make(chan BlockEvent, 16), - Attempts: make(chan AttemptEvent, 16), - Log: log, - PollWait: 500 * time.Millisecond, + Path: path, + Events: make(chan BlockEvent, 16), + Attempts: make(chan AttemptEvent, 16), + Latencies: make(chan LatencyEvent, 16), + Log: log, + PollWait: 500 * time.Millisecond, } } @@ -94,6 +104,9 @@ var ( // "Submitting possible block solve share diff N !" // "Possible remote block solve diff N !" solveDiffRE = regexp.MustCompile(`(?:Possible|Submitting[^"]*possible).*block solve.*diff\s+([0-9eE.+-]+)`) + // Matches the latency line from our patch 0005: + // "Block update latency: 92ms (ZMQ trigger to mining.notify broadcast)" + latencyRE = regexp.MustCompile(`Block update latency:\s+(\d+)ms`) ) // Run blocks until ctx is cancelled. It opens the file, seeks to either @@ -104,6 +117,7 @@ var ( func (t *Tailer) Run(ctx context.Context) { defer close(t.Events) defer close(t.Attempts) + defer close(t.Latencies) var ( f *os.File @@ -242,6 +256,15 @@ func (t *Tailer) Run(ctx context.Context) { } func (t *Tailer) handleLine(line string) { + if m := latencyRE.FindStringSubmatch(line); m != nil { + if ms, err := strconv.ParseInt(m[1], 10, 64); err == nil { + select { + case t.Latencies <- LatencyEvent{SeenAt: time.Now(), LatencyMs: ms}: + default: + } + } + return + } if m := solveDiffRE.FindStringSubmatch(line); m != nil { var d float64 if v, err := strconv.ParseFloat(m[1], 64); err == nil { diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go index 216353b..689cde2 100644 --- a/api/internal/state/aggregator.go +++ b/api/internal/state/aggregator.go @@ -88,6 +88,19 @@ type Snapshot struct { LastZMQEventAge float64 `json:"last_zmq_event_age,omitempty"` // seconds; >=0 HasLastZMQEvent bool `json:"has_last_zmq_event"` + // Share counters: raw counts (1 submission = 1 share regardless of diff). + // Session = since ckpool started; AllTime = persisted across restarts. + SessionAccepted int64 `json:"session_accepted"` + SessionRejected int64 `json:"session_rejected"` + AllTimeAccepted int64 `json:"alltime_accepted"` + AllTimeRejected int64 `json:"alltime_rejected"` + + // Block update latency diagnostics (ZMQ trigger → mining.notify). + LatencyCount int64 `json:"latency_count"` + LatencyAvgMs int64 `json:"latency_avg_ms"` + LatencyLastMs int64 `json:"latency_last_ms"` + StaleWorkHashes float64 `json:"stale_work_hashes"` + // Health CKPoolOK bool `json:"ckpool_ok"` BitcoinOK bool `json:"bitcoin_ok"` @@ -107,6 +120,14 @@ const ( kvSubmitAttempts = "block_submit_attempts" kvSubmitsConfirmed = "block_submits_confirmed" + kvLatencyCount = "latency_count" + kvLatencySumMs = "latency_sum_ms" + kvLatencyLastMs = "latency_last_ms" + kvStaleWorkHashes = "stale_work_hashes" + + kvAllTimeAccepted = "alltime_accepted" + kvAllTimeRejected = "alltime_rejected" + // reconcileInterval is how often we sweep recent blocks looking // for missing hash/reward enrichment and reorg-orphaned hashes. // 60s is fast enough to recover from a transient bitcoind hiccup @@ -192,6 +213,22 @@ type Aggregator struct { // from zmqmon. Used by /healthz to flag stale subscriptions. zmqEnabled bool lastZMQEventTime time.Time + + // All-time share counters (raw, not diff-weighted). Accumulated + // using the same delta-integration pattern as cumulative_shares. + allTimeAccepted int64 + allTimeRejected int64 + lastPoolAcceptedRaw int64 + lastPoolRejectedRaw int64 + hasShareCountBaseline bool + lastShareCountSave time.Time + + // Block update latency tracking (from ckpool patch 0005). + // Persisted to kv store so stats accumulate across restarts. + latencyCount int64 // number of observations + latencySumMs int64 // sum of all latencies in ms + latencyLastMs int64 // most recent observation + staleWorkHashes float64 // cumulative wasted hashes = sum(latency_s * hashrate_at_event) } func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator { @@ -396,6 +433,36 @@ func (a *Aggregator) refresh(ctx context.Context) { a.lastCumulativeSave = now } + // --- raw share count tracking (1 share = 1 submission) --- + // pool.Shares = raw accepted count, pool.RejectCount = raw rejected count. + // Delta-integrate the same way as cumulative work. + if next.Pool != nil { + curAcc := next.Pool.Shares + curRej := next.Pool.RejectCount + if a.hasShareCountBaseline { + if curAcc >= a.lastPoolAcceptedRaw { + a.allTimeAccepted += curAcc - a.lastPoolAcceptedRaw + } + if curRej >= a.lastPoolRejectedRaw { + a.allTimeRejected += curRej - a.lastPoolRejectedRaw + } + } + a.lastPoolAcceptedRaw = curAcc + a.lastPoolRejectedRaw = curRej + a.hasShareCountBaseline = true + next.SessionAccepted = curAcc + next.SessionRejected = curRej + } + next.AllTimeAccepted = a.allTimeAccepted + next.AllTimeRejected = a.allTimeRejected + + // Persist share counts at most once per minute. + if a.Store != nil && now.Sub(a.lastShareCountSave) >= time.Minute { + _ = a.Store.SetKV(kvAllTimeAccepted, strconv.FormatInt(a.allTimeAccepted, 10)) + _ = a.Store.SetKV(kvAllTimeRejected, strconv.FormatInt(a.allTimeRejected, 10)) + a.lastShareCountSave = now + } + // --- hashrate history sample (once per minute) --- if now.Sub(a.lastHRSampleAt) >= time.Minute { p := HashratePoint{T: now.Unix(), V: next.HashrateHs} @@ -455,6 +522,12 @@ func (a *Aggregator) refresh(ctx context.Context) { next.LastZMQEventAge = time.Since(a.lastZMQEventTime).Seconds() next.HasLastZMQEvent = true } + next.LatencyCount = a.latencyCount + if a.latencyCount > 0 { + next.LatencyAvgMs = a.latencySumMs / a.latencyCount + } + next.LatencyLastMs = a.latencyLastMs + next.StaleWorkHashes = a.staleWorkHashes a.snap = next cb := a.OnRefresh pushed := next @@ -500,6 +573,50 @@ func (a *Aggregator) loadPersistedState() { } } + // Restore all-time share counts. + if v, err := a.Store.GetKV(kvAllTimeAccepted); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.mu.Lock() + a.allTimeAccepted = n + a.mu.Unlock() + } + } + if v, err := a.Store.GetKV(kvAllTimeRejected); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.mu.Lock() + a.allTimeRejected = n + a.mu.Unlock() + } + } + + // Restore latency stats. + a.mu.Lock() + if v, err := a.Store.GetKV(kvLatencyCount); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.latencyCount = n + } + } + if v, err := a.Store.GetKV(kvLatencySumMs); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.latencySumMs = n + } + } + if v, err := a.Store.GetKV(kvLatencyLastMs); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.latencyLastMs = n + } + } + if v, err := a.Store.GetKV(kvStaleWorkHashes); err == nil && v != "" { + if f, perr := strconv.ParseFloat(v, 64); perr == nil { + a.staleWorkHashes = f + } + } + if a.latencyCount > 0 { + a.Log.Info("latency stats loaded", "count", a.latencyCount, + "avg_ms", a.latencySumMs/a.latencyCount, "last_ms", a.latencyLastMs) + } + a.mu.Unlock() + 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) diff --git a/api/internal/state/blocks.go b/api/internal/state/blocks.go index 16a9bad..f5a351b 100644 --- a/api/internal/state/blocks.go +++ b/api/internal/state/blocks.go @@ -80,6 +80,39 @@ func (a *Aggregator) IngestAttemptEvents(ctx context.Context, events <-chan logm } } +// IngestLatencyEvents reads block-update latency events from the tailer +// and accumulates stats for the dashboard. Stats are persisted to the +// kv store so they survive restarts. +func (a *Aggregator) IngestLatencyEvents(ctx context.Context, events <-chan logmon.LatencyEvent) { + for { + select { + case <-ctx.Done(): + return + case ev, ok := <-events: + if !ok { + return + } + a.mu.Lock() + a.latencyCount++ + a.latencySumMs += ev.LatencyMs + a.latencyLastMs = ev.LatencyMs + // Wasted hashes = latency in seconds × current 5m hashrate. + hs5m := a.snap.HashrateHs5m + a.staleWorkHashes += (float64(ev.LatencyMs) / 1000.0) * hs5m + count, sum, last, stale := a.latencyCount, a.latencySumMs, a.latencyLastMs, a.staleWorkHashes + a.mu.Unlock() + a.Log.Info("logmon: block update latency", "ms", ev.LatencyMs, "count", count) + // Blocks are infrequent (~10min); persist every event. + if a.Store != nil { + _ = a.Store.SetKV(kvLatencyCount, strconv.FormatInt(count, 10)) + _ = a.Store.SetKV(kvLatencySumMs, strconv.FormatInt(sum, 10)) + _ = a.Store.SetKV(kvLatencyLastMs, strconv.FormatInt(last, 10)) + _ = a.Store.SetKV(kvStaleWorkHashes, strconv.FormatFloat(stale, 'f', -1, 64)) + } + } + } +} + // IngestBlockEvents reads block events from the tailer and appends them // to the snapshot's block history. Runs until ctx is cancelled. func (a *Aggregator) IngestBlockEvents(ctx context.Context, events <-chan logmon.BlockEvent) { diff --git a/ckpool/patches/0005-log-zmq-to-notify-latency.patch b/ckpool/patches/0005-log-zmq-to-notify-latency.patch new file mode 100644 index 0000000..3e3649f --- /dev/null +++ b/ckpool/patches/0005-log-zmq-to-notify-latency.patch @@ -0,0 +1,38 @@ +diff --git a/src/stratifier.c b/src/stratifier.c +index a039aad3..d6355d6f 100644 +--- a/src/stratifier.c ++++ b/src/stratifier.c +@@ -429,6 +429,9 @@ struct stratifier_data { + /* Time we last sent out a stratum update */ + time_t update_time; + ++ /* Timestamp of last ZMQ block trigger for latency measurement */ ++ tv_t zmq_trigger_time; ++ + int64_t workbase_id; + int64_t blockchange_id; + int session_id; +@@ -1519,6 +1522,15 @@ retry: + stratum_broadcast_update(sdata, wb, new_block); + ret = true; + LOGINFO("Broadcast updated stratum base"); ++ if (new_block && sdata->zmq_trigger_time.tv_sec) { ++ tv_t now; ++ tv_time(&now); ++ LOGWARNING("Block update latency: %dms (ZMQ trigger to mining.notify broadcast)", ++ ms_tvdiff(&now, &sdata->zmq_trigger_time)); ++ } ++ /* Always clear ZMQ trigger so a stale timestamp from a reconnection ++ * or duplicate detection does not contaminate the next measurement. */ ++ sdata->zmq_trigger_time.tv_sec = 0; + /* Update transactions after stratum broadcast to not delay + * propagation. */ + if (likely(txns)) +@@ -8544,6 +8556,7 @@ static void *zmqnotify(void *arg) + LOGDEBUG("ZMQ sequence number"); + break; + case 32: ++ tv_time(&sdata->zmq_trigger_time); + update_base(sdata, GEN_PRIORITY); + __bin2hex(hexhash, zmq_msg_data(&message), 32); + LOGNOTICE("ZMQ block hash %s", hexhash); diff --git a/ckpool/patches/0006-expose-raw-reject-count-in-poolstats.patch b/ckpool/patches/0006-expose-raw-reject-count-in-poolstats.patch new file mode 100644 index 0000000..c8460dd --- /dev/null +++ b/ckpool/patches/0006-expose-raw-reject-count-in-poolstats.patch @@ -0,0 +1,69 @@ +diff --git a/src/stratifier.c b/src/stratifier.c +index d6355d6f..09b68097 100644 +--- a/src/stratifier.c ++++ b/src/stratifier.c +@@ -71,6 +71,8 @@ struct pool_stats { + int64_t accounted_diff_shares; + int64_t unaccounted_rejects; + int64_t accounted_rejects; ++ int64_t unaccounted_reject_count; ++ int64_t accounted_reject_count; + + /* Diff shares per second for 1/5/15... minute rolling averages */ + double dsps1; +@@ -4508,13 +4510,13 @@ static void get_poolstats(sdata_t *sdata, int *sockd) + json_t *val; + + mutex_lock(&sdata->stats_lock); +- JSON_CPACK(val, "{si,si,si,si,si,sI,sf,sf,sf,sf,sI,sI,sf,sf,sf,sf,sf,sf,sf}", ++ JSON_CPACK(val, "{si,si,si,si,si,sI,sf,sf,sf,sf,sI,sI,sI,sf,sf,sf,sf,sf,sf,sf}", + "start", stats->start_time.tv_sec, "update", stats->last_update.tv_sec, + "workers", stats->workers + stats->remote_workers, "users", stats->users + stats->remote_users, + "disconnected", stats->disconnected, + "shares", stats->accounted_shares, "sps1", stats->sps1, "sps5", stats->sps5, + "sps15", stats->sps15, "sps60", stats->sps60, "accepted", stats->accounted_diff_shares, +- "rejected", stats->accounted_rejects, "dsps1", stats->dsps1, "dsps5", stats->dsps5, ++ "rejected", stats->accounted_rejects, "rejectcount", stats->accounted_reject_count, "dsps1", stats->dsps1, "dsps5", stats->dsps5, + "dsps15", stats->dsps15, "dsps60", stats->dsps60, "dsps360", stats->dsps360, + "dsps1440", stats->dsps1440, "dsps10080", stats->dsps10080); + mutex_unlock(&sdata->stats_lock); +@@ -5625,8 +5627,10 @@ static void add_submit(ckpool_t *ckp, stratum_instance_t *client, const double d + if (valid) { + ckp_sdata->stats.unaccounted_shares++; + ckp_sdata->stats.unaccounted_diff_shares += diff; +- } else ++ } else { + ckp_sdata->stats.unaccounted_rejects += diff; ++ ckp_sdata->stats.unaccounted_reject_count++; ++ } + mutex_unlock(&ckp_sdata->uastats_lock); + + /* Count only accepted and stale rejects in diff calculation. */ +@@ -8325,7 +8329,7 @@ out_status: + for (i = 0; i < 32; i++) { + int64_t unaccounted_shares, + unaccounted_diff_shares, +- unaccounted_rejects; ++ unaccounted_rejects, unaccounted_reject_count; + + ts_to_tv(&diff, &stats->last_update); + cksleep_ms_r(&stats->last_update, 1875); +@@ -8339,15 +8343,18 @@ out_status: + unaccounted_shares = stats->unaccounted_shares; + unaccounted_diff_shares = stats->unaccounted_diff_shares; + unaccounted_rejects = stats->unaccounted_rejects; ++ unaccounted_reject_count = stats->unaccounted_reject_count; + stats->unaccounted_shares = + stats->unaccounted_diff_shares = + stats->unaccounted_rejects = 0; ++ stats->unaccounted_reject_count = 0; + mutex_unlock(&sdata->uastats_lock); + + mutex_lock(&sdata->stats_lock); + stats->accounted_shares += unaccounted_shares; + stats->accounted_diff_shares += unaccounted_diff_shares; + stats->accounted_rejects += unaccounted_rejects; ++ stats->accounted_reject_count += unaccounted_reject_count; + + decay_time(&stats->sps1, unaccounted_shares, per_tdiff, MIN1); + decay_time(&stats->sps5, unaccounted_shares, per_tdiff, MIN5); diff --git a/ui/src/App.svelte b/ui/src/App.svelte index c7ead45..3b4f296 100644 --- a/ui/src/App.svelte +++ b/ui/src/App.svelte @@ -11,6 +11,7 @@ import BestShares from "./lib/BestShares.svelte"; import UserDetailPage from "./lib/UserDetailPage.svelte"; import WorkerDetailPage from "./lib/WorkerDetailPage.svelte"; + import SharesBar from "./lib/SharesBar.svelte"; import BlockFoundAnimation from "./lib/BlockFoundAnimation.svelte"; onMount(() => { @@ -39,6 +40,7 @@ {:else} + diff --git a/ui/src/lib/MinersTable.svelte b/ui/src/lib/MinersTable.svelte index b2c0f14..38d88a7 100644 --- a/ui/src/lib/MinersTable.svelte +++ b/ui/src/lib/MinersTable.svelte @@ -96,7 +96,14 @@
-

Miners

+
+

Miners

+ + {rows.filter(r => r.online).length} online · + {rows.length} workers · + {new Set(rows.map(r => r.btcAddress)).size} users + +
{#if rows.length === 0}
No miners connected.
{:else} @@ -165,11 +172,21 @@
diff --git a/ui/src/lib/UserDetailPage.svelte b/ui/src/lib/UserDetailPage.svelte index 9385e7f..0405f7a 100644 --- a/ui/src/lib/UserDetailPage.svelte +++ b/ui/src/lib/UserDetailPage.svelte @@ -134,6 +134,7 @@ }; }); + const explorerBase = $derived( explorerBaseFor(snap.data?.chain?.chain, snap.data?.mempool_base_url), ); @@ -198,6 +199,7 @@ +

Workers

{#if workerRows.length === 0} diff --git a/ui/src/lib/WorkerDetailPage.svelte b/ui/src/lib/WorkerDetailPage.svelte index 262640f..bae192f 100644 --- a/ui/src/lib/WorkerDetailPage.svelte +++ b/ui/src/lib/WorkerDetailPage.svelte @@ -62,6 +62,7 @@ return (workerShares / poolShares) * 100; }); + function onKey(ev: KeyboardEvent): void { if (ev.key === "Escape") clearSelection(); } diff --git a/ui/src/types.ts b/ui/src/types.ts index 100cc00..b6b2620 100644 --- a/ui/src/types.ts +++ b/ui/src/types.ts @@ -10,6 +10,7 @@ export type PoolStats = { shares: number; accepted: number; rejected: number; + rejectcount: number; dsps1: number; dsps5: number; dsps15: number; @@ -140,6 +141,18 @@ export type Snapshot = { zmq_enabled: boolean; has_last_zmq_event: boolean; last_zmq_event_age?: number; // seconds + // Share counters (raw, 1 submission = 1 share). + session_accepted: number; + session_rejected: number; + alltime_accepted: number; + alltime_rejected: number; + + // Block update latency diagnostics (ZMQ → mining.notify). + latency_count: number; + latency_avg_ms: number; + latency_last_ms: number; + stale_work_hashes: number; + ckpool_ok: boolean; bitcoin_ok: boolean; last_error?: string;