Harden block-recording pipeline: P0 reliability fixes
Closes the silent-failure modes between "ckpool logs a solve" and "block correctly displayed": * Difficulty estimate matched mempool.space — the projection now uses (inEpoch + 1) intervals so it converges on Bitcoin Core's eventual retarget formula at end-of-epoch instead of undershooting by ~0.05– 0.10 % throughout. * Tailer resumes mid-log on restart — persists (inode, offset) to kv every EOF + on shutdown, and replays the unread tail next time. Any solve line written while kamado-api was down would previously be invisible forever. * Background reconcile loop (60 s) retries hash/reward enrichment for blocks the original RPC missed, so a transient bitcoind-index race no longer permanently leaves a block hashless. * Reorg detection: same loop compares each recent stored hash against getblockhash(height); a mismatch stamps orphaned_at. UI renders these strikethrough with a red "orphaned" tag instead of showing illusory rewards forever. * InsertBlock now reports whether a row was actually inserted; the caller WARN-logs duplicate-height ignores so a re-mined orphaned height can't disappear silently. * Submit-attempt vs confirmed counters surface failed submissions: every "Possible/Submitting block solve" log line increments block_submit_attempts; "Solved and confirmed" increments block_submits_confirmed. A growing gap means bitcoind is rejecting our submissions — previously invisible. * share_err patch refreshed against pinned ckpool source: added SE_NO_JOBID -> 21 and SE_WORKER_MISMATCH -> 24 mappings, kept SE_INVALID_NONCE2 in 20 (it's a malformed-input error, not low-diff). AxeOS users now see actionable Stratum codes instead of "unknown error". UI gets new orphaned_at + block_submit_attempts/confirmed fields on the snapshot type and a strikethrough-with-tag rendering for orphaned blocks in BlocksTable.
This commit is contained in:
satoshi
+200
-16
@@ -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))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user