Add block-update latency tracking, raw share counters, and shares bar UI

Log ZMQ→mining.notify latency in ckpool (patch 0005), expose a raw
reject counter (patch 0006), and surface both in the dashboard:

- Block latency card shows avg/last ms, wasted work, and block count
- SharesBar component shows session + all-time accepted/rejected with
  a Demon Slayer flame-slash animation on new shares
- Miners card info moved to MinersTable section header
- Latency stats and share counts persist across restarts via kv store
- Removed misleading pool-wide share stats from per-worker/user pages
  (ckpool doesn't expose per-user raw counts)

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,

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"`

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 {

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)

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) {