Fix Share stats disappearing regression

api/cmd/kamado-api/main.go

@@ -101,6 +101,13 @@ func main() {
 	zmq := zmqmon.New(cfg.BitcoinZMQBlock, log)
 	go zmq.Run(ctx)
 
+	// Restore persisted state before anything can mutate it. The log
+	// tailer's ingest goroutines below share these counters and persist
+	// them on their first event, so starting them against a half-loaded
+	// aggregator overwrites the accumulated history with a few seconds
+	// of new shares.
+	agg.LoadPersisted()
+
 	go agg.Run(ctx, zmq.Events)
 
 	// Background reconciliation: retry hash/reward enrichment for blocks

api/internal/httpapi/server.go

@@ -43,6 +43,7 @@ func (s *Server) Handler() http.Handler {
 	mux.HandleFunc("POST /api/admin/reset-latency", s.resetLatency)
 	mux.HandleFunc("POST /api/admin/ack-best", s.ackBest)
 	mux.HandleFunc("POST /api/admin/reset-ack-best", s.resetAckBest)
+	mux.HandleFunc("POST /api/admin/rebuild-share-stats", s.rebuildShareStats)
 	mux.HandleFunc("POST /api/accelerate", s.accelerate)
 	mux.HandleFunc("POST /api/accelerate/cancel", s.accelerateCancel)
 	mux.HandleFunc("POST /api/accelerate/max", s.accelerateMax)
@@ -223,6 +224,22 @@ func (s *Server) resetLatency(w http.ResponseWriter, _ *http.Request) {
 	writeJSON(w, http.StatusOK, map[string]any{"ok": true})
 }
 
+// rebuildShareStats recounts the all-time share statistics from the
+// ckpool log. Blocking and potentially slow (it reads the entire log),
+// so it is only ever reached by an explicit operator request.
+func (s *Server) rebuildShareStats(w http.ResponseWriter, _ *http.Request) {
+	adopted, scanned, err := s.Agg.RebuildShareStatsFromLog()
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, map[string]any{
+			"ok": false, "error": err.Error(),
+		})
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"ok": true, "adopted": adopted, "shares_in_log": scanned,
+	})
+}
+
 func (s *Server) ackBest(w http.ResponseWriter, _ *http.Request) {
 	s.Agg.AckBestDiff()
 	writeJSON(w, http.StatusOK, map[string]any{"ok": true})

api/internal/logmon/tailer.go

@@ -570,6 +570,66 @@ func FindBestPowData(path string) (best *PowData, bestAt time.Time, seen, unpars
 	return best, bestAt, seen, unparseable, err
 }
 
+// ShareStats is a full accounting of the accepted and rejected shares
+// recorded in a ckpool log, used to rebuild the all-time counters when
+// the stored ones are lost or truncated. Diffs holds each accepted
+// share's difficulty in log order so the caller can bucket them with
+// its own thresholds rather than duplicating that policy here.
+type ShareStats struct {
+	Accepted      int64
+	DiffSum       float64
+	Diffs         []float64
+	RejectReasons map[string]int64
+}
+
+// ScanShareStats reads an entire ckpool log and tallies every accepted
+// and rejected share it contains. On a busy pool this walks millions of
+// lines, so it is meant for explicit recovery, never for the startup
+// path.
+func ScanShareStats(path string) (*ShareStats, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	st := &ShareStats{RejectReasons: make(map[string]int64)}
+	err = scanLines(f, func(line string) {
+		if strings.Contains(line, "Accepted client") {
+			m := acceptedShareRE.FindStringSubmatch(line)
+			if m == nil {
+				return
+			}
+			d, perr := strconv.ParseFloat(m[1], 64)
+			if perr != nil {
+				return
+			}
+			st.Accepted++
+			st.DiffSum += d
+			st.Diffs = append(st.Diffs, d)
+			return
+		}
+		if !strings.Contains(line, "Rejected client") {
+			return
+		}
+		if m := rejectedDiffRE.FindStringSubmatch(line); m != nil {
+			st.RejectReasons[rejectKeyword(m[1])]++
+			return
+		}
+		if m := rejectedInvalRE.FindStringSubmatch(line); m != nil {
+			reason := strings.TrimSpace(m[1])
+			if reason == "" {
+				reason = "Unknown"
+			}
+			st.RejectReasons[reason]++
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+	return st, nil
+}
+
 // sleep returns false if ctx was cancelled during the wait.
 func sleep(ctx context.Context, d time.Duration) bool {
 	select {

api/internal/state/aggregator.go

@@ -7,6 +7,7 @@ package state
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"log/slog"
 	"strconv"
 	"sync"
@@ -371,6 +372,17 @@ type Aggregator struct {
 	sessionDiffSum       float64 // sum of share difficulties (session)
 	alltimeDiffSum       float64 // sum of share difficulties (alltime)
 	lastShareStatsSave   time.Time
+
+	// loadOnce guards LoadPersisted so main can complete the restore
+	// before starting the ingest goroutines, while Run stays safe to
+	// call on its own.
+	loadOnce sync.Once
+	// statsLoaded reports that the alltime share statistics have been
+	// read back from the store. Persisting before that would write the
+	// zero-valued in-memory counters over the accumulated history — the
+	// first share event triggers a save, so this is the difference
+	// between a restart and permanent data loss.
+	statsLoaded bool
 }
 
 func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator {
@@ -387,6 +399,19 @@ func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog
 	}
 }
 
+// LoadPersisted restores block history and counters from the store.
+// Run calls it, but callers should invoke it directly before starting
+// any ingest goroutine: those goroutines mutate the same counters and
+// persist them on their first event, so a restore still in flight would
+// be overwritten by — and then read back as — near-empty state. Safe to
+// call repeatedly; only the first call does the work.
+func (a *Aggregator) LoadPersisted() {
+	a.loadOnce.Do(func() {
+		a.loadPersistedBlocks()
+		a.loadPersistedState()
+	})
+}
+
 // Ready returns a channel that's closed once the aggregator has
 // completed its first refresh — used by main to delay HTTP serving
 // until /api/snapshot reflects real state instead of zeros.
@@ -407,8 +432,7 @@ func (a *Aggregator) Run(ctx context.Context, tipEvents <-chan zmqmon.TipEvent)
 	a.mu.Lock()
 	a.zmqEnabled = tipEvents != nil
 	a.mu.Unlock()
-	a.loadPersistedBlocks()
-	a.loadPersistedState()
+	a.LoadPersisted()
 	a.refresh(ctx)
 	t := time.NewTicker(a.Interval)
 	defer t.Stop()
@@ -926,6 +950,50 @@ func (a *Aggregator) IngestShareEvents(ctx context.Context, events <-chan logmon
 	}
 }
 
+// RebuildShareStatsFromLog recounts the all-time share statistics by
+// scanning the whole ckpool log, and adopts the result when it accounts
+// for more shares than the counters currently hold. Recovery for stored
+// stats that were lost or truncated: the log is the only other record of
+// per-share difficulty, and ckpool keeps its full history there.
+//
+// Deliberately never automatic — the scan walks every line of a log that
+// can hold millions, and it was a long startup scan that made the race
+// this repairs so easy to hit in the first place.
+func (a *Aggregator) RebuildShareStatsFromLog() (adopted bool, scanned int64, err error) {
+	if a.LogFilePath == "" {
+		return false, 0, errors.New("no ckpool log path configured")
+	}
+	st, err := logmon.ScanShareStats(a.LogFilePath)
+	if err != nil {
+		return false, 0, err
+	}
+
+	var dist [6]int64
+	for _, d := range st.Diffs {
+		dist[diffBucket(d)]++
+	}
+
+	a.mu.Lock()
+	current := a.alltimeDiffCount
+	if st.Accepted <= current {
+		a.mu.Unlock()
+		a.Log.Info("share stats rebuild: log has no more shares than stored, keeping stored",
+			"log_shares", st.Accepted, "stored_shares", current)
+		return false, st.Accepted, nil
+	}
+	a.alltimeDiffDist = dist
+	a.alltimeDiffCount = st.Accepted
+	a.alltimeDiffSum = st.DiffSum
+	a.alltimeRejectReasons = st.RejectReasons
+	a.statsLoaded = true
+	a.mu.Unlock()
+
+	a.persistShareStats()
+	a.Log.Warn("share stats rebuilt from ckpool log",
+		"shares", st.Accepted, "was", current)
+	return true, st.Accepted, nil
+}
+
 // ResetSessionShareStats clears session-level share statistics. Called
 // when a ckpool restart is detected (pool.Shares decreases).
 func (a *Aggregator) resetSessionShareStats() {
@@ -939,6 +1007,15 @@ func (a *Aggregator) persistShareStats() {
 	if a.Store == nil {
 		return
 	}
+	a.mu.RLock()
+	loaded := a.statsLoaded
+	a.mu.RUnlock()
+	if !loaded {
+		// Belt and braces alongside the load-before-ingest ordering in
+		// main: writing here would replace the accumulated history with
+		// whatever few shares have arrived since startup.
+		return
+	}
 	a.mu.Lock()
 	reasonsCopy := make(map[string]int64, len(a.alltimeRejectReasons))
 	for k, v := range a.alltimeRejectReasons {
@@ -1165,6 +1242,14 @@ func (a *Aggregator) loadPersistedState() {
 		}
 	}
 
+	// From here the alltime counters reflect the store, so persisting
+	// them can no longer erase history.
+	a.mu.Lock()
+	a.statsLoaded = true
+	loadedShares := a.alltimeDiffCount
+	a.mu.Unlock()
+	a.Log.Info("share stats loaded", "alltime_shares", loadedShares)
+
 	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/pow_test.go

@@ -205,3 +205,97 @@ func TestPowPairing_LowerDiffDoesNotReplace(t *testing.T) {
 		t.Fatalf("bestPow = %+v, want the 9000-diff record kept", a.bestPow)
 	}
 }
+
+// Reproduces the startup race that destroyed accumulated share
+// statistics: the ingest goroutine runs concurrently with the restore,
+// and its first event triggers a persist. If that persist is allowed to
+// run before the counters are loaded, it writes the zero-valued
+// in-memory histogram over the stored history — which the restore then
+// reads back, making the loss permanent.
+func TestShareStats_NotPersistedBeforeLoad(t *testing.T) {
+	s := openTempStore(t)
+	// Seed the store as a long-running pool would: 4M accepted shares.
+	const seeded = `{"b":[4000000,0,0,0,12,3],"n":4000015,"s":4.2e10}`
+	if err := s.SetKV(kvDiffDistAll, seeded); err != nil {
+		t.Fatal(err)
+	}
+
+	// An aggregator that has NOT loaded yet, as during startup.
+	a := New(nil, nil, time.Minute, discardLog())
+	a.Store = s
+	ingest(t, a, logmon.ShareEvent{SeenAt: time.Now(), Diff: 1234, Hash: "aa"})
+
+	v, err := s.GetKV(kvDiffDistAll)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if v != seeded {
+		t.Fatalf("history overwritten before load:\n got %s\nwant %s", v, seeded)
+	}
+
+	// After loading, the accumulated history is in memory and further
+	// shares extend it instead of replacing it.
+	a.LoadPersisted()
+	a.mu.RLock()
+	count := a.alltimeDiffCount
+	bucket0 := a.alltimeDiffDist[0]
+	a.mu.RUnlock()
+	if count != 4000015 || bucket0 != 4000000 {
+		t.Fatalf("loaded count=%d bucket0=%d, want 4000015/4000000", count, bucket0)
+	}
+
+	ingest(t, a, logmon.ShareEvent{SeenAt: time.Now(), Diff: 1234, Hash: "bb"})
+	a.mu.RLock()
+	after := a.alltimeDiffCount
+	a.mu.RUnlock()
+	if after != 4000016 {
+		t.Errorf("alltime count = %d after one more share, want 4000016", after)
+	}
+}
+
+// Recovery path: rebuild the all-time histogram by rescanning the log
+// when the stored counters hold fewer shares than the log accounts for.
+func TestRebuildShareStatsFromLog(t *testing.T) {
+	logPath := tempLogFile(t,
+		`[2026-09-02 10:00:00.001] Accepted client 1 share diff 500/42/1.0K: aa`+"\n",
+		`[2026-09-02 10:00:00.002] Accepted client 1 share diff 2000000/42/1.0K: bb`+"\n",
+		`[2026-09-02 10:00:00.003] Accepted client 1 share diff 500000000000/42/1.0K: cc`+"\n",
+		`[2026-09-02 10:00:00.004] Rejected client 1 dupe diff 500/42/1.0K: aa`+"\n",
+		`[2026-09-02 10:00:00.005] Rejected client 1 invalid share Stale`+"\n",
+	)
+	s := openTempStore(t)
+	a := New(nil, nil, time.Minute, discardLog())
+	a.Store = s
+	a.LogFilePath = logPath
+	a.LoadPersisted()
+
+	adopted, scanned, err := a.RebuildShareStatsFromLog()
+	if err != nil {
+		t.Fatalf("rebuild: %v", err)
+	}
+	if !adopted || scanned != 3 {
+		t.Fatalf("adopted=%v scanned=%d, want true/3", adopted, scanned)
+	}
+
+	a.mu.RLock()
+	dist := a.alltimeDiffDist
+	count := a.alltimeDiffCount
+	reasons := a.alltimeRejectReasons
+	a.mu.RUnlock()
+	// buckets: <1M, 1M-100M, 100M-1G, 1G-100G, 100G-1T, >=1T
+	if dist[0] != 1 || dist[1] != 1 || dist[4] != 1 || count != 3 {
+		t.Errorf("dist=%v count=%d, want one share each in buckets 0,1,4", dist, count)
+	}
+	if reasons["Duplicate"] != 1 || reasons["Stale"] != 1 {
+		t.Errorf("reject reasons = %v, want one Duplicate and one Stale", reasons)
+	}
+
+	// A rebuild that would shrink the history must be refused.
+	a.mu.Lock()
+	a.alltimeDiffCount = 4000000
+	a.mu.Unlock()
+	adopted, _, err = a.RebuildShareStatsFromLog()
+	if err != nil || adopted {
+		t.Errorf("adopted=%v err=%v, want refusal when the log holds fewer shares", adopted, err)
+	}
+}