diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go index 46ac697..3804e9e 100644 --- a/api/cmd/kamado-api/main.go +++ b/api/cmd/kamado-api/main.go @@ -8,10 +8,13 @@ package main import ( "context" "errors" + "fmt" "log/slog" "net/http" "os" "os/signal" + "strconv" + "strings" "syscall" "time" @@ -78,10 +81,40 @@ func main() { go agg.Run(ctx, zmq.Events) + // Background reconciliation: retry hash/reward enrichment for blocks + // the initial RPC lookup couldn't fetch, and detect chain reorgs by + // comparing recorded hashes against the canonical chain. + go agg.ReconcileBlocks(ctx) + // Tail the ckpool log for block-solve events (our own solves). tailer := logmon.New(cfg.CKPoolLogFile, log) + if blockStore != nil { + const cursorKey = "logmon_cursor" + tailer.LoadCursor = func() (uint64, int64, bool) { + v, err := blockStore.GetKV(cursorKey) + if err != nil || v == "" { + return 0, 0, false + } + parts := strings.SplitN(v, ":", 2) + if len(parts) != 2 { + return 0, 0, false + } + ino, err1 := strconv.ParseUint(parts[0], 10, 64) + off, err2 := strconv.ParseInt(parts[1], 10, 64) + if err1 != nil || err2 != nil { + return 0, 0, false + } + return ino, off, true + } + tailer.SaveCursor = func(ino uint64, off int64) { + if err := blockStore.SetKV(cursorKey, fmt.Sprintf("%d:%d", ino, off)); err != nil { + log.Warn("logmon cursor persist failed", "err", err) + } + } + } go tailer.Run(ctx) go agg.IngestBlockEvents(ctx, tailer.Events) + go agg.IngestAttemptEvents(ctx, tailer.Attempts) srv := &http.Server{ Addr: cfg.ListenAddr, diff --git a/api/internal/logmon/tailer.go b/api/internal/logmon/tailer.go index 2295a99..5cd0c26 100644 --- a/api/internal/logmon/tailer.go +++ b/api/internal/logmon/tailer.go @@ -36,24 +36,50 @@ type BlockEvent struct { ShareDiff float64 `json:"share_diff,omitempty"` } +// AttemptEvent is emitted when ckpool logs a "Possible block solve" +// or "Submitting possible block solve" line — i.e. ckpool decided a +// share met network difficulty and is attempting to submit it to +// bitcoind. This precedes the "Solved and confirmed" line that drives +// BlockEvent. Counting attempts vs confirmations surfaces submission +// failures (bitcoind rejected, RPC timeout, etc.) that would otherwise +// be invisible. +type AttemptEvent struct { + SeenAt time.Time + ShareDiff float64 + RawLine string +} + // 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 + // LoadCursor / SaveCursor, if both non-nil, persist the read + // position across process restarts. LoadCursor returns (inode, + // offset, true) when a saved cursor exists for this path; the + // tailer will resume from offset only if the inode still matches. + // SaveCursor is invoked on a throttled cadence as we read, so a + // crash loses at most ~1s of unread bytes. + LoadCursor func() (inode uint64, offset int64, ok bool) + SaveCursor func(inode uint64, offset int64) + // lastSolveDiff remembers the share diff from the most recent // "Possible block solve" line so handleLine can attach it to the // subsequent "Solved and confirmed block" event. Reset after use. lastSolveDiff float64 + + lastCursorSave time.Time } 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, } @@ -70,17 +96,24 @@ var ( solveDiffRE = regexp.MustCompile(`(?:Possible|Submitting[^"]*possible).*block solve.*diff\s+([0-9eE.+-]+)`) ) -// Run blocks until ctx is cancelled. It opens the file, seeks to the end, -// and reads new lines as they are appended. If the file is rotated -// (shrinks, or inode changes), it re-opens. +// 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 +// are appended. If the file is rotated (shrinks, or inode changes), it +// re-opens. func (t *Tailer) Run(ctx context.Context) { defer close(t.Events) + defer close(t.Attempts) var ( f *os.File reader *bufio.Reader lastIno uint64 lastPos int64 + // firstOpen distinguishes the very first Open of this tailer + // (where we honor LoadCursor and replay the backlog) from + // rotation re-opens (where we always start at 0). + firstOpen = true ) open := func() error { @@ -91,23 +124,64 @@ func (t *Tailer) Run(ctx context.Context) { if err != nil { return err } - // Start at end on first open so we don't replay old events. - if _, err := nf.Seek(0, io.SeekEnd); err != nil { - _ = nf.Close() - return err - } st, err := nf.Stat() if err != nil { _ = nf.Close() return err } + ino := inodeOf(st) + size := st.Size() + + // Decide where to start reading. + var startAt int64 + if firstOpen { + if t.LoadCursor != nil { + if savedIno, savedOff, ok := t.LoadCursor(); ok && + savedIno == ino && savedOff <= size { + startAt = savedOff + if savedOff < size { + t.Log.Info("logmon: resuming from saved cursor", + "path", t.Path, "offset", savedOff, "size", size) + } + } else { + // No cursor, or stale (file was rotated since we + // last saw it, so we have no way to know what we + // already read). Start at end to avoid replaying + // the entire log. + startAt = size + } + } else { + startAt = size + } + firstOpen = false + } else { + // Rotation: read the whole replacement file from the start. + startAt = 0 + } + + if _, err := nf.Seek(startAt, io.SeekStart); err != nil { + _ = nf.Close() + return err + } f = nf reader = bufio.NewReader(f) - lastIno = inodeOf(st) - lastPos, _ = f.Seek(0, io.SeekCurrent) + lastIno = ino + lastPos = startAt return nil } + saveCursor := func(force bool) { + if t.SaveCursor == nil || f == nil { + return + } + now := time.Now() + if !force && now.Sub(t.lastCursorSave) < time.Second { + return + } + t.SaveCursor(lastIno, lastPos) + t.lastCursorSave = now + } + // Initial open; retry on failure until the file exists. for { if err := open(); err != nil { @@ -127,6 +201,7 @@ func (t *Tailer) Run(ctx context.Context) { for { if ctx.Err() != nil { + saveCursor(true) return } line, err := reader.ReadString('\n') @@ -139,6 +214,7 @@ func (t *Tailer) Run(ctx context.Context) { } if !errors.Is(err, io.EOF) { t.Log.Warn("logmon: read error, reopening", "err", err) + saveCursor(true) if !sleep(ctx, t.PollWait) { return } @@ -146,7 +222,9 @@ func (t *Tailer) Run(ctx context.Context) { continue } - // EOF: check for rotation (inode changed) or truncation (size < pos). + // EOF: persist where we are (caught up to current end), then + // check for rotation (inode changed) or truncation (size < pos). + saveCursor(false) if st, statErr := os.Stat(t.Path); statErr == nil { if inodeOf(st) != lastIno || st.Size() < lastPos { t.Log.Info("logmon: log rotated, reopening", "path", t.Path) @@ -164,8 +242,15 @@ func (t *Tailer) Run(ctx context.Context) { func (t *Tailer) handleLine(line string) { if m := solveDiffRE.FindStringSubmatch(line); m != nil { - if d, err := strconv.ParseFloat(m[1], 64); err == nil { - t.lastSolveDiff = d + var d float64 + if v, err := strconv.ParseFloat(m[1], 64); err == nil { + d = v + t.lastSolveDiff = v + } + select { + case t.Attempts <- AttemptEvent{SeenAt: time.Now(), ShareDiff: d, RawLine: line}: + default: + // Best-effort metric — dropping is fine. } return } diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go index 13e8550..c6d3469 100644 --- a/api/internal/state/aggregator.go +++ b/api/internal/state/aggregator.go @@ -73,6 +73,13 @@ type Snapshot struct { // the StartOS config "Block Explorer" -> "Custom URL". MempoolBaseURL string `json:"mempool_base_url,omitempty"` + // Counts of share-submit attempts ("Possible/Submitting block solve" + // log lines) and confirmed solves ("Solved and confirmed block"). + // A growing gap means bitcoind is rejecting our submissions or + // dropping the RPC — surface it in the UI as an alert. + BlockSubmitAttempts int64 `json:"block_submit_attempts"` + BlockSubmitsConfirmed int64 `json:"block_submits_confirmed"` + // Health CKPoolOK bool `json:"ckpool_ok"` BitcoinOK bool `json:"bitcoin_ok"` @@ -88,6 +95,20 @@ const ( // diff-1-normalized work, so cumulative_work * 2^32 is real hashes. // The old key is orphaned in the kv table on upgrade; harmless. kvCumulativeWork = "cumulative_work" + + kvSubmitAttempts = "block_submit_attempts" + kvSubmitsConfirmed = "block_submits_confirmed" + + // 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 + // within a minute, slow enough to never load the RPC. + reconcileInterval = 60 * time.Second + + // reconcileLookback caps how far back the reconcile loop looks. + // Blocks older than this with empty hash are abandoned; hashes + // older than this are assumed deep enough to never reorg. + reconcileLookback = 24 * time.Hour ) // Aggregator refreshes a Snapshot on a ticker. @@ -146,6 +167,12 @@ type Aggregator struct { // DEBUG (likely a transient warm-up or lock hiccup); repeated failures // escalate to WARN. ckFailStreak int + + // Submit-attempt vs confirmed counters. Persisted in kv so the + // running gap survives restarts. Both are monotonic. + blockSubmitAttempts int64 + blockSubmitsConfirmed int64 + lastSubmitCountSave time.Time } func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator { @@ -270,7 +297,12 @@ func (a *Aggregator) refresh(ctx context.Context) { } if a.retargetStartUnix > 0 { elapsed := float64(time.Now().Unix() - a.retargetStartUnix) - expected := float64(inEpoch) * 600 + // Match Bitcoin Core / mempool.space: project nActualTimespan + // by treating the elapsed window as covering (inEpoch + 1) + // block intervals, since at retarget the consensus formula + // uses (lastBlock.time - firstBlock.time) across all 2016 + // blocks of the epoch. + expected := float64(inEpoch+1) * 600 if elapsed > 0 { factor := expected / elapsed if factor > 4 { @@ -379,6 +411,8 @@ func (a *Aggregator) refresh(ctx context.Context) { next.RecentBlocks = make([]BlockRecord, len(a.blocks)) copy(next.RecentBlocks, a.blocks) } + next.BlockSubmitAttempts = a.blockSubmitAttempts + next.BlockSubmitsConfirmed = a.blockSubmitsConfirmed a.snap = next cb := a.OnRefresh pushed := next @@ -408,6 +442,21 @@ func (a *Aggregator) loadPersistedState() { } } + if v, err := a.Store.GetKV(kvSubmitAttempts); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.mu.Lock() + a.blockSubmitAttempts = n + a.mu.Unlock() + } + } + if v, err := a.Store.GetKV(kvSubmitsConfirmed); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.mu.Lock() + a.blockSubmitsConfirmed = n + 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 1df615d..00c7ffe 100644 --- a/api/internal/state/blocks.go +++ b/api/internal/state/blocks.go @@ -2,6 +2,7 @@ package state import ( "context" + "strconv" "time" "github.com/kamadopool/kamado-api/internal/logmon" @@ -10,14 +11,49 @@ import ( // BlockRecord is a found block, merged from a logmon event with bitcoind // data if available. Hash and Reward are populated best-effort via the -// RPC lookup scheduled right after the log line is seen. +// RPC lookup scheduled right after the log line is seen; the reconcile +// loop fills any holes later. OrphanedAt is set if a periodic chain +// check finds the recorded hash no longer matches the canonical block +// at this height (i.e. the network reorged us out). type BlockRecord struct { - Height int64 `json:"height"` - Hash string `json:"hash,omitempty"` - RewardBT float64 `json:"reward_btc,omitempty"` - FoundAt time.Time `json:"found_at"` - Source string `json:"source"` // "logmon" for now; "zmq" later - ShareDiff float64 `json:"share_diff,omitempty"` + Height int64 `json:"height"` + Hash string `json:"hash,omitempty"` + RewardBT float64 `json:"reward_btc,omitempty"` + FoundAt time.Time `json:"found_at"` + Source string `json:"source"` // "logmon" for now; "zmq" later + ShareDiff float64 `json:"share_diff,omitempty"` + OrphanedAt time.Time `json:"orphaned_at,omitempty"` +} + +// IngestAttemptEvents counts "Possible/Submitting block solve" log +// lines so we can compare attempts vs confirmations in the snapshot. +// A growing gap means bitcoind is rejecting our submissions or the +// RPC is failing. Persists the running counter so it survives restarts. +func (a *Aggregator) IngestAttemptEvents(ctx context.Context, events <-chan logmon.AttemptEvent) { + for { + select { + case <-ctx.Done(): + return + case ev, ok := <-events: + if !ok { + return + } + a.mu.Lock() + a.blockSubmitAttempts++ + n := a.blockSubmitAttempts + save := a.Store != nil && time.Since(a.lastSubmitCountSave) >= 30*time.Second + if save { + a.lastSubmitCountSave = time.Now() + } + a.mu.Unlock() + a.Log.Info("logmon: submit attempt", "share_diff", ev.ShareDiff, "attempts", n) + if save { + if err := a.Store.SetKV(kvSubmitAttempts, strconv.FormatInt(n, 10)); err != nil { + a.Log.Warn("submit attempts persist failed", "err", err) + } + } + } + } } // IngestBlockEvents reads block events from the tailer and appends them @@ -56,20 +92,38 @@ func (a *Aggregator) IngestBlockEvents(ctx context.Context, events <-chan logmon } cancel() } + isNew := true if a.Store != nil { - if err := a.Store.InsertBlock(store.Block{ + inserted, err := a.Store.InsertBlock(store.Block{ Height: rec.Height, Hash: rec.Hash, RewardBT: rec.RewardBT, FoundAt: rec.FoundAt, Source: rec.Source, ShareDiff: rec.ShareDiff, - }); err != nil { + }) + if err != nil { a.Log.Warn("block persist failed", "height", rec.Height, "err", err) + } else if !inserted { + a.Log.Warn("block at this height already recorded — duplicate or pre-existing entry, not counting as new", + "height", rec.Height, "hash", rec.Hash) + isNew = false + } + } + if !isNew { + continue + } + a.mu.Lock() + a.blockSubmitsConfirmed++ + confirmed := a.blockSubmitsConfirmed + a.mu.Unlock() + if a.Store != nil { + if err := a.Store.SetKV(kvSubmitsConfirmed, strconv.FormatInt(confirmed, 10)); err != nil { + a.Log.Warn("confirmed count persist failed", "err", err) } } pushed := a.appendBlock(rec) - a.Log.Info("block recorded", "height", rec.Height, "hash", rec.Hash) + a.Log.Info("block recorded", "height", rec.Height, "hash", rec.Hash, "confirmed", confirmed) // Push immediately so WebSocket clients see the solve // without waiting for the next poll tick. if a.OnRefresh != nil { @@ -79,6 +133,119 @@ func (a *Aggregator) IngestBlockEvents(ctx context.Context, events <-chan logmon } } +// ReconcileBlocks runs until ctx is cancelled, periodically: +// 1. Filling in missing hash/reward for blocks where the initial RPC +// lookup failed (bitcoind hadn't indexed yet, or was down). +// 2. Comparing each non-orphaned hash against the canonical block at +// its height; a mismatch means the network reorged us out and we +// mark the row orphaned so the UI can render it accordingly. +// Both checks are bounded to the last reconcileLookback, so the cost +// stays constant regardless of total history size. +func (a *Aggregator) ReconcileBlocks(ctx context.Context) { + if a.Store == nil || a.RPC == nil { + return + } + t := time.NewTicker(reconcileInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + a.reconcileOnce(ctx) + } + } +} + +func (a *Aggregator) reconcileOnce(ctx context.Context) { + since := time.Now().Add(-reconcileLookback) + + // Pass 1: enrichment. Fetch hash + reward for any missing-data rows. + missing, err := a.Store.BlocksNeedingEnrichment(since) + if err != nil { + a.Log.Warn("reconcile: load missing failed", "err", err) + } + enrichedAny := false + for _, b := range missing { + lookupCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + hash := b.Hash + reward := b.RewardBT + if hash == "" { + if h, herr := a.RPC.GetBlockHash(lookupCtx, b.Height); herr == nil { + hash = h + } else { + cancel() + continue + } + } + if reward == 0 && hash != "" { + if blk, berr := a.RPC.GetBlock(lookupCtx, hash); berr == nil { + reward = blk.CoinbaseReward() + } + } + cancel() + if hash != b.Hash || reward != b.RewardBT { + if err := a.Store.UpdateEnrichment(b.Height, hash, reward); err != nil { + a.Log.Warn("reconcile: update enrichment failed", "height", b.Height, "err", err) + continue + } + a.Log.Info("reconcile: enriched block", "height", b.Height, "hash", hash, "reward", reward) + enrichedAny = true + } + } + + // Pass 2: reorg detection. For non-orphaned blocks within the + // lookback, confirm the canonical hash at that height still + // matches our record. Don't bother with rows that are already + // orphaned — we won't un-orphan, since the network already + // chose another chain. + recent, err := a.Store.Recent(64) + if err != nil { + a.Log.Warn("reconcile: load recent failed", "err", err) + return + } + orphanedAny := false + now := time.Now() + for _, b := range recent { + if b.Hash == "" || !b.OrphanedAt.IsZero() { + continue + } + if b.FoundAt.Before(since) { + continue + } + lookupCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + canonical, herr := a.RPC.GetBlockHash(lookupCtx, b.Height) + cancel() + if herr != nil { + // Most likely cause: our bitcoind doesn't have this + // height yet (index lag). Try again next sweep — don't + // orphan on a transient RPC error. + continue + } + if canonical != b.Hash { + if err := a.Store.MarkOrphaned(b.Height, now); err != nil { + a.Log.Warn("reconcile: mark orphaned failed", "height", b.Height, "err", err) + continue + } + a.Log.Warn("reconcile: block orphaned by reorg", + "height", b.Height, "ours", b.Hash, "canonical", canonical) + orphanedAny = true + } + } + + if enrichedAny || orphanedAny { + // Refresh the in-memory ring so the snapshot picks up the + // changes immediately rather than waiting for the next poll. + a.loadPersistedBlocks() + a.mu.RLock() + snap := a.snap + a.mu.RUnlock() + if a.OnRefresh != nil && len(snap.RecentBlocks) > 0 { + a.OnRefresh(snap) + } + } +} + // maxBlockHistory caps in-memory block history. Persistence comes in // Phase 2b.5 via SQLite; for now recent blocks survive only this // process's lifetime. @@ -117,16 +284,33 @@ func (a *Aggregator) loadPersistedBlocks() { for i := len(rows) - 1; i >= 0; i-- { r := rows[i] recs = append(recs, BlockRecord{ - Height: r.Height, - Hash: r.Hash, - RewardBT: r.RewardBT, - FoundAt: r.FoundAt, - Source: r.Source, - ShareDiff: r.ShareDiff, + Height: r.Height, + Hash: r.Hash, + RewardBT: r.RewardBT, + FoundAt: r.FoundAt, + Source: r.Source, + ShareDiff: r.ShareDiff, + OrphanedAt: r.OrphanedAt, }) } a.mu.Lock() a.blocks = recs + // Surface refreshed history into the live snapshot so /api/snapshot + // reflects the latest store state without waiting for the next + // refresh tick (used by the reconcile loop). + if a.snap.GeneratedAt.IsZero() { + a.mu.Unlock() + a.Log.Info("block history loaded", "count", len(recs)) + return + } + snap := a.snap + if len(recs) > 0 { + snap.RecentBlocks = make([]BlockRecord, len(recs)) + copy(snap.RecentBlocks, recs) + } else { + snap.RecentBlocks = nil + } + a.snap = snap a.mu.Unlock() a.Log.Info("block history loaded", "count", len(recs)) } diff --git a/api/internal/store/blocks.go b/api/internal/store/blocks.go index e53a8bd..4a35156 100644 --- a/api/internal/store/blocks.go +++ b/api/internal/store/blocks.go @@ -21,22 +21,24 @@ type BlockStore struct { // package is the lower layer; the state package converts to/from this // shape when reading and writing. type Block struct { - Height int64 - Hash string - RewardBT float64 - FoundAt time.Time - Source string - ShareDiff float64 + Height int64 + Hash string + RewardBT float64 + FoundAt time.Time + Source string + ShareDiff float64 + OrphanedAt time.Time // zero value = not orphaned } const schema = ` CREATE TABLE IF NOT EXISTS blocks ( - height INTEGER PRIMARY KEY, - hash TEXT NOT NULL DEFAULT '', - reward_btc REAL NOT NULL DEFAULT 0, - found_at INTEGER NOT NULL, - source TEXT NOT NULL DEFAULT '', - share_diff REAL NOT NULL DEFAULT 0 + height INTEGER PRIMARY KEY, + hash TEXT NOT NULL DEFAULT '', + reward_btc REAL NOT NULL DEFAULT 0, + found_at INTEGER NOT NULL, + source TEXT NOT NULL DEFAULT '', + share_diff REAL NOT NULL DEFAULT 0, + orphaned_at INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS blocks_found_at_idx ON blocks(found_at); @@ -76,6 +78,11 @@ func (s *BlockStore) migrate() error { return fmt.Errorf("store: add share_diff: %w", err) } } + if !have["orphaned_at"] { + if _, err := s.db.Exec(`ALTER TABLE blocks ADD COLUMN orphaned_at INTEGER NOT NULL DEFAULT 0`); err != nil { + return fmt.Errorf("store: add orphaned_at: %w", err) + } + } return nil } @@ -104,13 +111,25 @@ func (s *BlockStore) Close() error { // InsertBlock is idempotent — duplicate heights are ignored so replayed // log events after a restart don't trip the primary key constraint. -func (s *BlockStore) InsertBlock(b Block) error { - _, err := s.db.Exec( +// Returns (true, nil) if a new row was actually inserted, (false, nil) +// if the height was already present (replay or duplicate). +// +// Limitation: the schema PK is height alone, so a self-mined block at +// the same height as a previously-orphaned self-mined block at that +// height (vanishingly unlikely for a solo pool) would also be dropped. +// Callers should log the (false, nil) case loudly so this never goes +// unnoticed. +func (s *BlockStore) InsertBlock(b Block) (bool, error) { + res, err := s.db.Exec( `INSERT OR IGNORE INTO blocks(height, hash, reward_btc, found_at, source, share_diff) VALUES (?, ?, ?, ?, ?, ?)`, b.Height, b.Hash, b.RewardBT, b.FoundAt.Unix(), b.Source, b.ShareDiff, ) - return err + if err != nil { + return false, err + } + n, err := res.RowsAffected() + return n > 0, err } // HashratePoint is one persisted hashrate sample. @@ -182,7 +201,7 @@ func (s *BlockStore) Recent(limit int) ([]Block, error) { limit = 256 } rows, err := s.db.Query( - `SELECT height, hash, reward_btc, found_at, source, share_diff + `SELECT height, hash, reward_btc, found_at, source, share_diff, orphaned_at FROM blocks ORDER BY height DESC LIMIT ?`, limit, ) @@ -193,11 +212,14 @@ func (s *BlockStore) Recent(limit int) ([]Block, error) { out := make([]Block, 0, limit) for rows.Next() { var b Block - var unix int64 - if err := rows.Scan(&b.Height, &b.Hash, &b.RewardBT, &unix, &b.Source, &b.ShareDiff); err != nil { + var foundUnix, orphanedUnix int64 + if err := rows.Scan(&b.Height, &b.Hash, &b.RewardBT, &foundUnix, &b.Source, &b.ShareDiff, &orphanedUnix); err != nil { return nil, err } - b.FoundAt = time.Unix(unix, 0).UTC() + b.FoundAt = time.Unix(foundUnix, 0).UTC() + if orphanedUnix > 0 { + b.OrphanedAt = time.Unix(orphanedUnix, 0).UTC() + } out = append(out, b) } if err := rows.Err(); err != nil && !errors.Is(err, sql.ErrNoRows) { @@ -205,3 +227,56 @@ func (s *BlockStore) Recent(limit int) ([]Block, error) { } return out, nil } + +// BlocksNeedingEnrichment returns blocks whose hash or reward is still +// unset and that were found within the lookback window. Older entries +// are ignored — if bitcoind couldn't tell us about a day-old block, +// retrying will keep failing. +func (s *BlockStore) BlocksNeedingEnrichment(since time.Time) ([]Block, error) { + rows, err := s.db.Query( + `SELECT height, hash, reward_btc, found_at, source, share_diff, orphaned_at + FROM blocks + WHERE found_at >= ? AND (hash = '' OR reward_btc = 0) + ORDER BY height ASC`, + since.Unix(), + ) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Block + for rows.Next() { + var b Block + var foundUnix, orphanedUnix int64 + if err := rows.Scan(&b.Height, &b.Hash, &b.RewardBT, &foundUnix, &b.Source, &b.ShareDiff, &orphanedUnix); err != nil { + return nil, err + } + b.FoundAt = time.Unix(foundUnix, 0).UTC() + if orphanedUnix > 0 { + b.OrphanedAt = time.Unix(orphanedUnix, 0).UTC() + } + out = append(out, b) + } + return out, rows.Err() +} + +// UpdateEnrichment fills in hash and reward for an already-recorded +// block. No-op if the row doesn't exist. +func (s *BlockStore) UpdateEnrichment(height int64, hash string, reward float64) error { + _, err := s.db.Exec( + `UPDATE blocks SET hash = ?, reward_btc = ? WHERE height = ?`, + hash, reward, height, + ) + return err +} + +// MarkOrphaned stamps a block as reorged-out at the given time. The +// found_at + reward fields stay so the UI can still render it with a +// strikethrough. +func (s *BlockStore) MarkOrphaned(height int64, at time.Time) error { + _, err := s.db.Exec( + `UPDATE blocks SET orphaned_at = ? WHERE height = ?`, + at.Unix(), height, + ) + return err +} diff --git a/ckpool/patches/0003-share-error-as-stratum-array.patch b/ckpool/patches/0003-share-error-as-stratum-array.patch index 9b36a2e..ba5c41e 100644 --- a/ckpool/patches/0003-share-error-as-stratum-array.patch +++ b/ckpool/patches/0003-share-error-as-stratum-array.patch @@ -1,8 +1,8 @@ diff --git a/src/stratifier.c b/src/stratifier.c -index 8281fa0..47d0fed 100644 +index 8281fa0..52da790 100644 --- a/src/stratifier.c +++ b/src/stratifier.c -@@ -6004,7 +6004,30 @@ static void check_best_diff(sdata_t *sdata, user_instance_t *user,worker_instanc +@@ -6004,7 +6004,32 @@ static void check_best_diff(sdata_t *sdata, user_instance_t *user,worker_instanc stratum_send_message(sdata, client, buf); } @@ -15,8 +15,9 @@ index 8281fa0..47d0fed 100644 +static inline int share_err_code(enum share_err err) +{ + switch (err) { -+ case SE_STALE: ++ case SE_NO_JOBID: + case SE_INVALID_JOBID: ++ case SE_STALE: + case SE_NTIME_INVALID: + return 21; /* job not found / stale */ + case SE_DUPE: @@ -24,6 +25,7 @@ index 8281fa0..47d0fed 100644 + case SE_HIGH_DIFF: + return 23; /* low difficulty share */ + case SE_NO_USERNAME: ++ case SE_WORKER_MISMATCH: + return 24; /* unauthorized worker */ + default: + return 20; /* other / unknown */ diff --git a/ui/src/lib/BlocksTable.svelte b/ui/src/lib/BlocksTable.svelte index 13324eb..a1dd739 100644 --- a/ui/src/lib/BlocksTable.svelte +++ b/ui/src/lib/BlocksTable.svelte @@ -30,8 +30,13 @@ {#each blocks as b (b.height + "-" + b.found_at)} - - {b.height} + + + {b.height} + {#if b.orphaned_at} + orphaned + {/if} + {#if b.hash}