diff --git a/README.md b/README.md index 53ccb2b..fdff51d 100644 --- a/README.md +++ b/README.md @@ -95,6 +95,7 @@ The Svelte 5 dashboard connects via WebSocket for real-time push updates (no pol - **ckpool kill-on-failure** — When bitcoind becomes unreachable, kamado-api kills ckpool so miners can failover to another pool. ckpool restarts automatically when bitcoind recovers. - **Deferred RPC** — Dashboard RPCs wait until the ckpool notifier completes on tip change to prioritize system resources for block reconstruction and validation. - **Log cursor persistence** — The tailer's file offset is stored in SQLite so block-solve detection resumes correctly after restart. +- **Load before ingest** — Persisted counters are restored *before* the log-tailer goroutines start. They share the same state and persist it on their first event, so starting them against a half-restored aggregator would write a few seconds of new shares over the accumulated all-time history — and then read that back as the new truth. `RebuildShareStatsFromLog` (exposed as `POST /api/admin/rebuild-share-stats`) recounts the all-time difficulty distribution and rejection reasons from the ckpool log if they are ever lost; it only adopts the result when the log accounts for more shares than the stored totals, so it can restore history but never erase it. - **Clean restarts for miners** — Upstream ckpool sends a reconnecting miner its first job *before* telling it the difficulty, and resets every client to the pool's `startdiff`, discarding what vardiff had tuned. Both produce a burst of "Above target" rejects on every pool restart. Patch `0009` sends `mining.set_difficulty` first, seeds the starting difficulty from the worker's retained hashrate, and accepts shares for the already-in-flight job down to the pool minimum until the next `mining.notify`. - **Warm-up resilience** — The ckpool socket client handles transient EOFs during startup gracefully instead of crashing. diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go index aa08736..3ff12eb 100644 --- a/api/cmd/kamado-api/main.go +++ b/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 diff --git a/api/internal/httpapi/server.go b/api/internal/httpapi/server.go index b060bcd..7266d62 100644 --- a/api/internal/httpapi/server.go +++ b/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}) diff --git a/api/internal/logmon/tailer.go b/api/internal/logmon/tailer.go index 9534806..3302e16 100644 --- a/api/internal/logmon/tailer.go +++ b/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 { diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go index e7539f5..8723da7 100644 --- a/api/internal/state/aggregator.go +++ b/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) diff --git a/api/internal/state/pow_test.go b/api/internal/state/pow_test.go index beabd6b..bad5295 100644 --- a/api/internal/state/pow_test.go +++ b/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) + } +}