Add a page which proves the best share PoW

api/internal/logmon/tailer.go

@@ -15,6 +15,7 @@ package logmon
 import (
 	"bufio"
 	"context"
+	"encoding/json"
 	"errors"
 	"io"
 	"log/slog"
@@ -67,6 +68,31 @@ type ShareEvent struct {
 	Hash     string  // block header hash (hex, accepted shares only)
 	Rejected bool
 	Reason   string // rejection reason (e.g. "Stale", "Duplicate"); empty for accepted
+	// Pow is set instead of the fields above when the line was a
+	// "Best share PoW data" record (ckpool patch 0008). It travels on
+	// the same channel as accepted/rejected events so ordering with
+	// the matching "Accepted client" line is preserved.
+	Pow *PowData
+}
+
+// PowData is the proof-of-work reproduction record ckpool logs (patch
+// 0008) when a share sets a new process-wide best difficulty: the raw
+// 80-byte block header exactly as it was double-SHA256'd, the assembled
+// coinbase transaction, and the merkle branch hashes linking that
+// coinbase to the merkle root committed in the header. Field names
+// mirror the JSON emitted by the patch.
+type PowData struct {
+	SDiff          float64  `json:"sdiff"`
+	NetDiff        float64  `json:"netdiff"`
+	Height         int64    `json:"height"`
+	Hash           string   `json:"hash"`     // display-order (big-endian) block header hash
+	Header         string   `json:"header"`   // 160 hex chars, hashing byte order
+	Coinbase       string   `json:"coinbase"` // full serialized coinbase tx, hex
+	CB1Len         int      `json:"cb1len"`   // bytes preceding extranonce1 in the coinbase
+	MerkleBranches []string `json:"merklebranches"`
+	Enonce1        string   `json:"enonce1"`
+	Nonce2         string   `json:"nonce2"`
+	Workername     string   `json:"workername"`
 }
 
 // Tailer follows a log file, surviving rotation/truncation, and emits
@@ -136,6 +162,40 @@ var (
 	rejectedInvalRE = regexp.MustCompile(`Rejected client \S+ invalid share (.+)`)
 )
 
+// powDataMarker precedes the JSON payload of a PoW reproduction line
+// from our patch 0008, logged just before the matching "Accepted client"
+// line when a share sets a new process-wide best:
+//
+//	"Best share PoW data {"sdiff":...,"header":"...",...}"
+const powDataMarker = "Best share PoW data "
+
+// parsePowLine extracts the PoW record from a log line carrying
+// powDataMarker. It decodes exactly one JSON value starting at the first
+// '{' after the marker and deliberately ignores everything after it —
+// production logs have carried trailing bytes after the closing brace,
+// which silently defeated an end-of-line-anchored regex. Returns
+// (nil, nil) when the marker is absent, and a non-nil error for a
+// present-but-undecodable record.
+func parsePowLine(line string) (*PowData, error) {
+	i := strings.Index(line, powDataMarker)
+	if i < 0 {
+		return nil, nil
+	}
+	rest := line[i+len(powDataMarker):]
+	j := strings.IndexByte(rest, '{')
+	if j < 0 {
+		return nil, errors.New("no JSON object after PoW data marker")
+	}
+	var pd PowData
+	if err := json.NewDecoder(strings.NewReader(rest[j:])).Decode(&pd); err != nil {
+		return nil, err
+	}
+	if pd.Header == "" || pd.Hash == "" {
+		return nil, errors.New("PoW record missing header or hash")
+	}
+	return &pd, nil
+}
+
 // Run blocks until ctx is cancelled. It opens the file, seeks to either
 // the saved cursor (if LoadCursor returns one and the inode still
 // matches) or the end on first-ever run, and reads new lines as they
@@ -284,6 +344,22 @@ func (t *Tailer) Run(ctx context.Context) {
 }
 
 func (t *Tailer) handleLine(line string) {
+	// PoW reproduction data (patch 0008). Must ship on the Shares
+	// channel so it stays ordered before its "Accepted client" line.
+	if strings.Contains(line, powDataMarker) {
+		pd, err := parsePowLine(line)
+		if err != nil {
+			t.Log.Warn("logmon: unparseable PoW data line", "err", err)
+			return
+		}
+		if pd != nil {
+			select {
+			case t.Shares <- ShareEvent{SeenAt: time.Now(), Pow: pd}:
+			default:
+			}
+		}
+		return
+	}
 	// Accepted share: extract share difficulty and block header hash.
 	if m := acceptedShareRE.FindStringSubmatch(line); m != nil {
 		if d, err := strconv.ParseFloat(m[1], 64); err == nil {
@@ -384,29 +460,116 @@ func FindBestShareHash(path string) (float64, string) {
 
 	var bestDiff float64
 	var bestHash string
-	scanner := bufio.NewScanner(f)
-	scanner.Buffer(make([]byte, 0, 64*1024), 256*1024)
-	for scanner.Scan() {
-		line := scanner.Text()
+	_ = scanLines(f, func(line string) {
 		if !strings.Contains(line, "Accepted client") {
-			continue
+			return
 		}
 		m := acceptedShareRE.FindStringSubmatch(line)
 		if m == nil {
-			continue
+			return
 		}
 		d, err := strconv.ParseFloat(m[1], 64)
 		if err != nil {
-			continue
+			return
 		}
 		if d >= bestDiff {
 			bestDiff = d
 			bestHash = m[2]
 		}
-	}
+	})
 	return bestDiff, bestHash
 }
 
+// scanLines invokes fn for every line of r. Unlike bufio.Scanner, a line
+// of any length can never abort the scan: an oversized line is passed to
+// fn truncated to the buffer size and its remainder is discarded. A
+// months-old production log with one pathological line must not silently
+// stop a backfill scan partway through — that failure mode is invisible
+// and permanent.
+func scanLines(f *os.File, fn func(line string)) error {
+	r := bufio.NewReaderSize(f, 256*1024)
+	skipping := false
+	for {
+		chunk, isPrefix, err := r.ReadLine()
+		if len(chunk) > 0 && !skipping {
+			fn(string(chunk))
+		}
+		skipping = isPrefix
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return nil
+			}
+			return err
+		}
+	}
+}
+
+// logStampRE extracts the "[YYYY-MM-DD HH:MM:SS.mmm]" prefix ckpool
+// puts on every log line (local time of the ckpool process).
+var logStampRE = regexp.MustCompile(`^\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})`)
+
+// FindBestPowData scans the log for the highest-difficulty "Best share
+// PoW data" record (patch 0008) whose share was subsequently confirmed
+// by an "Accepted client" line with the same hash. Used as a one-time
+// backfill when kamado-api holds no persisted record but the log
+// already contains some (fresh database, or PoW events dropped under
+// load) — without it the PoW inspector would sit empty until a share
+// beat ckpool's in-process best, which after a long uptime can take
+// arbitrarily long. The returned time is parsed from the log line's
+// timestamp; zero when unparseable. Returns a nil record if no confirmed
+// record exists; err reports open/read failures and the returned counts
+// say how many PoW lines were seen and how many failed to parse, so the
+// caller can log a diagnosable outcome instead of a silent nil.
+func FindBestPowData(path string) (best *PowData, bestAt time.Time, seen, unparseable int, err error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, time.Time{}, 0, 0, err
+	}
+	defer f.Close()
+
+	// Records awaiting their "Accepted client" confirmation, keyed by
+	// hash. Bounded the same way as the aggregator's live pairing.
+	type parked struct {
+		pow *PowData
+		at  time.Time
+	}
+	pending := make(map[string]parked)
+	err = scanLines(f, func(line string) {
+		if strings.Contains(line, powDataMarker) {
+			seen++
+			pd, perr := parsePowLine(line)
+			if perr != nil || pd == nil {
+				unparseable++
+				return
+			}
+			var at time.Time
+			if sm := logStampRE.FindStringSubmatch(line); sm != nil {
+				at, _ = time.ParseInLocation("2006-01-02 15:04:05", sm[1], time.Local)
+			}
+			if len(pending) >= 16 {
+				pending = make(map[string]parked)
+			}
+			pending[pd.Hash] = parked{pow: pd, at: at}
+			return
+		}
+		if !strings.Contains(line, "Accepted client") {
+			return
+		}
+		m := acceptedShareRE.FindStringSubmatch(line)
+		if m == nil {
+			return
+		}
+		if p, ok := pending[m[2]]; ok {
+			delete(pending, m[2])
+			if best == nil || p.pow.SDiff > best.SDiff {
+				best = p.pow
+				bestAt = p.at
+			}
+		}
+	})
+	return best, bestAt, seen, unparseable, err
+}
+
 // sleep returns false if ctx was cancelled during the wait.
 func sleep(ctx context.Context, d time.Duration) bool {
 	select {

api/internal/logmon/tailer_test.go

@@ -5,6 +5,8 @@ import (
 	"io"
 	"log/slog"
 	"os"
+	"strconv"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -118,6 +120,158 @@ func TestHandleLine_Unrelated(t *testing.T) {
 	}
 }
 
+func TestHandleLine_PowData(t *testing.T) {
+	tl := New("/unused", discardLog())
+	line := `[2026-08-31 10:11:12.123] Best share PoW data {"sdiff":123456789.5,"netdiff":136597951737045.0,"height":911111,` +
+		`"hash":"0000000000abcdef0000000000000000000000000000000000000000deadbeef",` +
+		`"header":"20000000" + "aa",` + "\n"
+	// Malformed JSON must not emit an event.
+	tl.handleLine(line)
+	select {
+	case ev := <-tl.Shares:
+		t.Errorf("unexpected ShareEvent from malformed PoW line: %+v", ev)
+	default:
+	}
+
+	good := `[2026-08-31 10:11:12.123] Best share PoW data {"sdiff":123456789.5,"netdiff":136597951737045.0,"height":911111,` +
+		`"hash":"0000000000abcdef0000000000000000000000000000000000000000deadbeef",` +
+		`"header":"0000002006226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f00000000000000000000000000000000000000000000000000000000000000006859a06817e0377c00000000",` +
+		`"coinbase":"01000000010000000000000000000000000000000000000000000000000000000000000000ffffffff00ffffffff0100f2052a01000000160014000000000000000000000000000000000000000000000000",` +
+		`"cb1len":41,"merklebranches":["ab"],"enonce1":"a1b2c3d4","nonce2":"0011223344556677","workername":"bc1qexample.axe"}` + "\n"
+	tl.handleLine(good)
+	select {
+	case ev := <-tl.Shares:
+		if ev.Pow == nil {
+			t.Fatalf("ShareEvent.Pow = nil, want parsed PoW data (event %+v)", ev)
+		}
+		if ev.Pow.SDiff != 123456789.5 {
+			t.Errorf("SDiff = %v, want 123456789.5", ev.Pow.SDiff)
+		}
+		if ev.Pow.Height != 911111 {
+			t.Errorf("Height = %d, want 911111", ev.Pow.Height)
+		}
+		if len(ev.Pow.Header) != 160 {
+			t.Errorf("Header length = %d, want 160", len(ev.Pow.Header))
+		}
+		if ev.Pow.CB1Len != 41 {
+			t.Errorf("CB1Len = %d, want 41", ev.Pow.CB1Len)
+		}
+		if len(ev.Pow.MerkleBranches) != 1 || ev.Pow.MerkleBranches[0] != "ab" {
+			t.Errorf("MerkleBranches = %v, want [ab]", ev.Pow.MerkleBranches)
+		}
+		if ev.Pow.Workername != "bc1qexample.axe" {
+			t.Errorf("Workername = %q", ev.Pow.Workername)
+		}
+	default:
+		t.Fatal("no ShareEvent emitted for PoW data line")
+	}
+}
+
+func powLine(hash string, sdiff float64) string {
+	return `[2026-08-30 09:15:42.123] Best share PoW data {"sdiff":` +
+		strconv.FormatFloat(sdiff, 'f', -1, 64) +
+		`,"netdiff":1.3e14,"height":911111,"hash":"` + hash +
+		`","header":"00000020aabb","coinbase":"0100","cb1len":41,` +
+		`"merklebranches":[],"enonce1":"a1b2c3d4","nonce2":"0011223344556677","workername":"bc1q.axe"}` + "\n"
+}
+
+func acceptedLine(hash string, sdiff float64) string {
+	return `[2026-08-30 09:15:42.124] Accepted client 7 share diff ` +
+		strconv.FormatFloat(sdiff, 'f', -1, 64) + `/42/1.0K: ` + hash + "\n"
+}
+
+// FindBestPowData must return the highest-diff record that was
+// confirmed accepted, skip unconfirmed ones, and recover the log
+// line's timestamp.
+func TestFindBestPowData(t *testing.T) {
+	path := tempLog(t)
+	appendLines(t, path,
+		powLine("aaaa", 100),
+		acceptedLine("aaaa", 100),
+		// Never confirmed (e.g. rejected below vardiff): must be ignored
+		// even though it carries the highest diff.
+		powLine("bbbb", 99999),
+		powLine("cccc", 500),
+		acceptedLine("cccc", 500),
+	)
+
+	pow, at, seen, unparseable, err := FindBestPowData(path)
+	if err != nil {
+		t.Fatalf("scan error: %v", err)
+	}
+	if pow == nil {
+		t.Fatal("FindBestPowData returned nil, want the confirmed 500-diff record")
+	}
+	if pow.Hash != "cccc" || pow.SDiff != 500 {
+		t.Errorf("got hash %s sdiff %v, want cccc/500", pow.Hash, pow.SDiff)
+	}
+	if seen != 3 || unparseable != 0 {
+		t.Errorf("seen=%d unparseable=%d, want 3/0", seen, unparseable)
+	}
+	want := time.Date(2026, 8, 30, 9, 15, 42, 0, time.Local)
+	if !at.Equal(want) {
+		t.Errorf("timestamp = %v, want %v", at, want)
+	}
+}
+
+// Production ckpool logs have carried bytes after the JSON's closing
+// brace (observed in the field: an end-of-line-anchored regex matched 0
+// of 54 real PoW lines while synthetic lines passed). The parser must
+// decode the JSON value and ignore ANY trailing bytes — NULs, \r,
+// or an interleaved second message on the same line.
+func TestHandleLine_PowDataTrailingGarbage(t *testing.T) {
+	suffixes := []string{
+		"\x00\x00\n",  // NUL padding
+		"\r\n",        // CR before LF
+		" [2026-08-31 18:00:01.125] Accepted client 3 share diff 1.0/1/1: aa\n", // interleaved write
+	}
+	for _, suffix := range suffixes {
+		tl := New("/unused", discardLog())
+		tl.handleLine(strings.TrimRight(powLine("cafe", 777), "\n") + suffix)
+		select {
+		case ev := <-tl.Shares:
+			if ev.Pow == nil || ev.Pow.Hash != "cafe" || ev.Pow.SDiff != 777 {
+				t.Errorf("suffix %q: got %+v, want cafe/777 PoW event", suffix, ev.Pow)
+			}
+		default:
+			t.Errorf("suffix %q: no PoW event emitted", suffix)
+		}
+	}
+}
+
+// A pathologically long line earlier in the log must not abort the scan
+// before it reaches the PoW records — bufio.Scanner did exactly that,
+// silently, which left the backfill permanently empty on real logs.
+func TestFindBestPowData_SurvivesOversizedLine(t *testing.T) {
+	path := tempLog(t)
+	appendLines(t, path,
+		"junk "+strings.Repeat("x", 600*1024)+"\n", // > any fixed buffer
+		powLine("abab", 250),
+		acceptedLine("abab", 250),
+	)
+	pow, _, seen, _, err := FindBestPowData(path)
+	if err != nil {
+		t.Fatalf("scan error: %v", err)
+	}
+	if pow == nil || pow.Hash != "abab" || seen != 1 {
+		t.Fatalf("pow=%+v seen=%d, want abab record found past the oversized line", pow, seen)
+	}
+}
+
+func TestFindBestPowData_Empty(t *testing.T) {
+	path := tempLog(t)
+	appendLines(t, path,
+		"Stratifier started\n",
+		acceptedLine("dddd", 10), // accepted share without PoW data (pre-patch line)
+	)
+	if pow, _, _, _, err := FindBestPowData(path); pow != nil || err != nil {
+		t.Errorf("got %+v err=%v, want nil/nil for log without confirmed PoW records", pow, err)
+	}
+	if pow, _, _, _, err := FindBestPowData(path + ".missing"); pow != nil || err == nil {
+		t.Errorf("got %+v err=%v, want nil record and an error for missing file", pow, err)
+	}
+}
+
 // ---- Tailer.Run integration tests (filesystem + goroutines) ----
 
 func tempLog(t *testing.T) string {