diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go index 98f56a7..fc56a6a 100644 --- a/api/cmd/kamado-api/main.go +++ b/api/cmd/kamado-api/main.go @@ -112,16 +112,10 @@ func main() { } } } - go tailer.Run(ctx) - go agg.IngestBlockEvents(ctx, tailer.Events) - go agg.IngestAttemptEvents(ctx, tailer.Attempts) - - // Wait briefly for the aggregator's first refresh to complete so - // the very first /api/snapshot or /api/health hit doesn't see an - // all-zeros snapshot and report bitcoin_ok=false during its own - // initialization. Cap the wait so a permanently-down bitcoind - // can't block startup forever — /healthz is honest about the - // degraded state. + // Wait for the aggregator's first refresh before starting the log + // tailer. This guarantees currentChain is set before any block + // events are processed, so blocks are always stamped with the + // correct chain — preventing false orphans on network switches. select { case <-agg.Ready(): log.Info("first snapshot ready") @@ -131,6 +125,10 @@ func main() { return } + go tailer.Run(ctx) + go agg.IngestBlockEvents(ctx, tailer.Events) + go agg.IngestAttemptEvents(ctx, tailer.Attempts) + srv := &http.Server{ Addr: cfg.ListenAddr, Handler: api.Handler(), diff --git a/api/internal/httpapi/server.go b/api/internal/httpapi/server.go index 32fd23d..b48063a 100644 --- a/api/internal/httpapi/server.go +++ b/api/internal/httpapi/server.go @@ -37,6 +37,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /api/blocks", s.blocks) mux.HandleFunc("GET /api/snapshot", s.snapshot) mux.HandleFunc("GET /api/ws", s.handleWS) + mux.HandleFunc("GET /api/admin/debug-blocks", s.debugBlocks) mux.Handle("/", spaHandler(webui.FS())) return mux } @@ -159,3 +160,51 @@ func (s *Server) clients(w http.ResponseWriter, r *http.Request) { func (s *Server) blocks(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, s.Agg.Blocks()) } + +// debugBlocks exposes raw block state from both the in-memory snapshot +// (what the UI sees) and the DB (ground truth) for troubleshooting. +func (s *Server) debugBlocks(w http.ResponseWriter, r *http.Request) { + type row struct { + Height int64 `json:"height"` + Hash string `json:"hash"` + Chain string `json:"chain"` + OrphanedAt string `json:"orphaned_at"` + FoundAt string `json:"found_at"` + RewardBTC float64 `json:"reward_btc"` + Source string `json:"source"` + } + fmtBlock := func(b state.BlockRecord) row { + orphaned := "" + if b.OrphanedAt != nil { + orphaned = b.OrphanedAt.Format("2006-01-02T15:04:05Z") + } + return row{ + Height: b.Height, + Hash: b.Hash, + Chain: b.Chain, + OrphanedAt: orphaned, + FoundAt: b.FoundAt.Format("2006-01-02T15:04:05Z"), + RewardBTC: b.RewardBT, + Source: b.Source, + } + } + + // In-memory blocks (what /api/blocks and /api/snapshot serve) + memBlocks := s.Agg.Blocks() + mem := make([]row, 0, len(memBlocks)) + for _, b := range memBlocks { + mem = append(mem, fmtBlock(b)) + } + + // DB blocks (ground truth) + dbBlocks := s.Agg.BlocksFromStore() + db := make([]row, 0, len(dbBlocks)) + for _, b := range dbBlocks { + db = append(db, fmtBlock(b)) + } + + writeJSON(w, http.StatusOK, map[string]any{ + "memory": mem, + "db": db, + }) +} diff --git a/api/internal/state/blocks.go b/api/internal/state/blocks.go index 598351b..16a9bad 100644 --- a/api/internal/state/blocks.go +++ b/api/internal/state/blocks.go @@ -16,14 +16,37 @@ import ( // 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"` - OrphanedAt time.Time `json:"orphaned_at,omitempty"` - Chain string `json:"chain,omitempty"` // "main", "test", "signet" + 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"` + Chain string `json:"chain,omitempty"` // "main", "test", "signet" +} + +// timePtr returns a pointer to t if non-zero, nil otherwise. +func timePtr(t time.Time) *time.Time { + if t.IsZero() { + return nil + } + return &t +} + +// inferOtherChain returns the most likely chain name for a block that +// doesn't belong to currentChain. Since the pool only supports mainnet +// and testnet4, the inference is unambiguous. Note: Bitcoin Core reports +// testnet4 as "testnet4" (not "test" which is the deprecated testnet3). +func inferOtherChain(currentChain string) string { + switch currentChain { + case "main": + return "testnet4" + case "testnet4": + return "main" + default: + return "other" + } } // IngestAttemptEvents counts "Possible/Submitting block solve" log @@ -171,13 +194,31 @@ func (a *Aggregator) ReconcileBlocks(ctx context.Context) { func (a *Aggregator) reconcileOnce(ctx context.Context) { since := time.Now().Add(-reconcileLookback) + // We need the current chain for both enrichment and reorg detection. + // If the first refresh hasn't completed yet, currentChain is "" and + // any chain comparison would be meaningless — skip until the next tick. + a.mu.RLock() + currentChain := "" + if a.snap.Chain != nil { + currentChain = a.snap.Chain.Chain + } + a.mu.RUnlock() + if currentChain == "" { + return + } + // Pass 1: enrichment. Fetch hash + reward for any missing-data rows. + // Skip blocks from a different chain — enriching a testnet block + // against mainnet RPC would overwrite its hash with the wrong value. missing, err := a.Store.BlocksNeedingEnrichment(since) if err != nil { a.Log.Warn("reconcile: load missing failed", "err", err) } enrichedAny := false for _, b := range missing { + if b.Chain != "" && b.Chain != currentChain { + continue + } lookupCtx, cancel := context.WithTimeout(ctx, 5*time.Second) hash := b.Hash reward := b.RewardBT @@ -205,29 +246,28 @@ func (a *Aggregator) reconcileOnce(ctx context.Context) { } } - // 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. - // - // Skip blocks whose recorded chain doesn't match the current - // bitcoind network — they were mined on a different node config - // (e.g. testnet) and cannot be checked against mainnet's tip. - // A mismatch here is a false positive, not a real reorg. - a.mu.RLock() - currentChain := "" - if a.snap.Chain != nil { - currentChain = a.snap.Chain.Chain - } - a.mu.RUnlock() - recent, err := a.Store.Recent(64) if err != nil { a.Log.Warn("reconcile: load recent failed", "err", err) return } + + // Pass 2: reorg detection + cross-network identification. + // + // For each non-orphaned block within the lookback that has a hash: + // 1. If its Chain field already marks it as a different network, skip. + // 2. Compare our hash against the canonical hash at that height. + // 3. On mismatch, call GetBlock(our_hash) to check if the block + // exists anywhere on the current chain (even if reorged out). + // - If GetBlock succeeds: it's a genuine reorg → mark orphaned. + // - If GetBlock fails ("Block not found"): the block doesn't + // exist on this network at all → it's from a different chain. + // Stamp it and leave it non-orphaned. + // + // This handles both stamped blocks (Chain="test") and legacy blocks + // (Chain="") that were recorded before the chain-stamp feature. orphanedAny := false + stampedAny := false now := time.Now() for _, b := range recent { if b.Hash == "" || !b.OrphanedAt.IsZero() { @@ -236,36 +276,149 @@ func (a *Aggregator) reconcileOnce(ctx context.Context) { if b.FoundAt.Before(since) { continue } - // Both chain values must be known before we can compare them. - // Legacy rows (Chain=="") fall through so existing installs keep - // their reorg detection — there was no network switch before this - // feature landed. - if b.Chain != "" && currentChain != "" && b.Chain != currentChain { - continue + if b.Chain != "" && b.Chain != currentChain { + continue // already known to be from a different network } 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 + continue // RPC transient error — retry next sweep } - 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) + if canonical == b.Hash { + continue // matches the canonical chain — all good + } + // Hash mismatch. Determine whether it's a real reorg or a + // cross-network block by checking if our hash exists on this chain. + // After a real reorg, bitcoind still has the stale block in its + // store; a cross-network hash won't exist at all. + lookupCtx2, cancel2 := context.WithTimeout(ctx, 3*time.Second) + _, berr := a.RPC.GetBlock(lookupCtx2, b.Hash) + cancel2() + if berr != nil { + // Block not found on this network → cross-network block. + // Stamp its chain so future sweeps skip it immediately. + otherChain := b.Chain + if otherChain == "" { + otherChain = inferOtherChain(currentChain) + } + if err := a.Store.StampChain(b.Height, otherChain); err != nil { + a.Log.Warn("reconcile: stamp cross-network block 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 + a.Log.Info("reconcile: block belongs to a different network, not a reorg", + "height", b.Height, "hash", b.Hash, "block_chain", otherChain, "current_chain", currentChain) + stampedAny = true + continue + } + // Block exists on this network but isn't canonical → genuine reorg. + 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 + } + + // Pass 3: un-orphan blocks that were previously false-positived. + // This covers two cases: + // a) Blocks with Chain != currentChain that got orphaned before this + // fix was deployed — un-orphan them now. + // b) Same-chain blocks whose hash now matches the canonical chain + // (e.g. the "reorg" reverted before we checked again). + fixedAny := false + for _, b := range recent { + if b.OrphanedAt.IsZero() || b.Hash == "" { + continue + } + if b.FoundAt.Before(since) { + continue + } + if b.Chain != "" && b.Chain != currentChain { + // Cross-network orphan from before the fix — un-orphan. + if err := a.Store.UnmarkOrphaned(b.Height); err != nil { + a.Log.Warn("reconcile: unmark cross-network orphan failed", "height", b.Height, "err", err) + continue + } + a.Log.Info("reconcile: cleared false orphan (different network)", + "height", b.Height, "block_chain", b.Chain, "current_chain", currentChain) + fixedAny = true + continue + } + // For same-chain or legacy blocks: verify via RPC. + lookupCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + _, berr := a.RPC.GetBlock(lookupCtx, b.Hash) + cancel() + if berr != nil { + // Block doesn't exist on this chain — cross-network. + otherChain := b.Chain + if otherChain == "" || otherChain == currentChain { + otherChain = inferOtherChain(currentChain) + } + if err := a.Store.UnmarkOrphanedStampChain(b.Height, otherChain); err != nil { + a.Log.Warn("reconcile: unmark+stamp cross-network orphan failed", "height", b.Height, "err", err) + continue + } + a.Log.Info("reconcile: cleared false orphan (block not on this network)", + "height", b.Height, "hash", b.Hash, "stamped_chain", otherChain) + fixedAny = true + continue + } + // Block exists on this chain — check if it's canonical again. + lookupCtx2, cancel2 := context.WithTimeout(ctx, 3*time.Second) + canonical, herr := a.RPC.GetBlockHash(lookupCtx2, b.Height) + cancel2() + if herr != nil { + continue + } + if canonical == b.Hash { + if err := a.Store.UnmarkOrphaned(b.Height); err != nil { + a.Log.Warn("reconcile: unmark reverted-reorg failed", "height", b.Height, "err", err) + continue + } + a.Log.Info("reconcile: cleared orphan — hash matches canonical again", + "height", b.Height, "hash", b.Hash) + fixedAny = true } } - if enrichedAny || orphanedAny { - // Refresh the in-memory ring so the snapshot picks up the - // changes immediately rather than waiting for the next poll. + // Pass 4: stamp chain on legacy/mis-stamped blocks that have a hash. + // Handles Chain="" (never stamped), "other" (old fallback), and "test" + // (wrong identifier — testnet4 reports as "testnet4", not "test"). + // Uses GetBlock(hash) to determine if the block belongs to the current + // network or a different one. No lookback restriction — this is a + // one-time migration for pre-existing rows. + for _, b := range recent { + if (b.Chain != "" && b.Chain != "other" && b.Chain != "test") || b.Hash == "" { + continue // already properly stamped or no hash to check + } + lookupCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + _, berr := a.RPC.GetBlock(lookupCtx, b.Hash) + cancel() + if berr != nil { + // Block not found on this network — infer the other chain. + inferredChain := inferOtherChain(currentChain) + if err := a.Store.StampChain(b.Height, inferredChain); err != nil { + a.Log.Warn("reconcile: stamp legacy block failed", "height", b.Height, "err", err) + continue + } + a.Log.Info("reconcile: stamped legacy block as other-network", + "height", b.Height, "hash", b.Hash, "inferred_chain", inferredChain) + stampedAny = true + } else { + // Block exists on this network + if err := a.Store.StampChain(b.Height, currentChain); err != nil { + a.Log.Warn("reconcile: stamp legacy block failed", "height", b.Height, "err", err) + continue + } + a.Log.Info("reconcile: stamped legacy block as current chain", + "height", b.Height, "chain", currentChain) + stampedAny = true + } + } + + if enrichedAny || orphanedAny || stampedAny || fixedAny { a.loadPersistedBlocks() a.mu.RLock() snap := a.snap @@ -320,7 +473,7 @@ func (a *Aggregator) loadPersistedBlocks() { FoundAt: r.FoundAt, Source: r.Source, ShareDiff: r.ShareDiff, - OrphanedAt: r.OrphanedAt, + OrphanedAt: timePtr(r.OrphanedAt), Chain: r.Chain, }) } @@ -354,3 +507,31 @@ func (a *Aggregator) Blocks() []BlockRecord { copy(out, a.blocks) return out } + +// BlocksFromStore reads blocks directly from the DB (bypassing in-memory +// cache) for debugging discrepancies between memory and store. +func (a *Aggregator) BlocksFromStore() []BlockRecord { + if a.Store == nil { + return nil + } + rows, err := a.Store.Recent(maxBlockHistory) + if err != nil { + return nil + } + out := make([]BlockRecord, 0, len(rows)) + for i := len(rows) - 1; i >= 0; i-- { + r := rows[i] + out = append(out, BlockRecord{ + Height: r.Height, + Hash: r.Hash, + RewardBT: r.RewardBT, + FoundAt: r.FoundAt, + Source: r.Source, + ShareDiff: r.ShareDiff, + OrphanedAt: timePtr(r.OrphanedAt), + Chain: r.Chain, + }) + } + return out +} + diff --git a/api/internal/state/blocks_test.go b/api/internal/state/blocks_test.go index 0d1c77e..fdb8eb9 100644 --- a/api/internal/state/blocks_test.go +++ b/api/internal/state/blocks_test.go @@ -108,10 +108,15 @@ func TestReconcileOnce_ChainFilter_NoFalseOrphan(t *testing.T) { t.Fatal(err) } - // Mock RPC returns a DIFFERENT hash for height 100 — mainnet canonical. + // Mock RPC: getblockhash returns a different mainnet hash; getblock + // for our testnet hash returns "Block not found" — it doesn't exist + // on mainnet at all, confirming it's cross-network not a reorg. rpc := mockRPC(t, func(method string) (any, *rpcErrResp) { - if method == "getblockhash" { + switch method { + case "getblockhash": return "mainnet_hash_bbb", nil + case "getblock": + return nil, &rpcErrResp{Code: -5, Message: "Block not found"} } return nil, nil }) @@ -150,9 +155,17 @@ func TestReconcileOnce_GenuineReorg(t *testing.T) { t.Fatal(err) } + // getblockhash returns a different canonical hash (reorg happened). + // getblock for our hash SUCCEEDS — the block still exists in bitcoind's + // block store after a reorg, it's just no longer on the active chain. + // This distinguishes a real reorg from a cross-network block. rpc := mockRPC(t, func(method string) (any, *rpcErrResp) { - if method == "getblockhash" { - return "canonical_bbbb", nil // different — reorg happened + switch method { + case "getblockhash": + return "canonical_bbbb", nil + case "getblock": + // Block exists (reorged but still in store) + return map[string]any{"hash": "our_hash_aaaa", "height": 200, "confirmations": -1}, nil } return nil, nil }) @@ -215,7 +228,7 @@ func TestReconcileOnce_RPCError_NoOrphan(t *testing.T) { } } -// TestReconcileOnce_LegacyBlock_StillChecked verifies that a legacy block +// TestReconcileOnce_LegacyBlock_GenuineReorg verifies that a legacy block // (Chain == "") is still checked for reorgs on the current chain, preserving // backward compatibility for installs that existed before chain was tracked. func TestReconcileOnce_LegacyBlock_StillChecked(t *testing.T) { @@ -230,10 +243,15 @@ func TestReconcileOnce_LegacyBlock_StillChecked(t *testing.T) { Chain: "", // legacy — no chain recorded }) - // Mainnet canonical hash differs: should orphan even though Chain is empty. + // Canonical hash differs AND getblock succeeds (block exists on this + // chain but was reorged out). Should still orphan legacy blocks. rpc := mockRPC(t, func(method string) (any, *rpcErrResp) { - if method == "getblockhash" { + switch method { + case "getblockhash": return "different_canonical", nil + case "getblock": + // Block exists in this chain's store — genuine reorg + return map[string]any{"hash": "legacy_hash", "height": 400, "confirmations": -1}, nil } return nil, nil }) @@ -251,6 +269,51 @@ func TestReconcileOnce_LegacyBlock_StillChecked(t *testing.T) { } } +// TestReconcileOnce_LegacyBlock_CrossNetwork verifies that a legacy block +// (Chain == "") whose hash doesn't exist on the current chain is identified +// as cross-network and stamped rather than orphaned. +func TestReconcileOnce_LegacyBlock_CrossNetwork(t *testing.T) { + st := openTempStore(t) + + st.InsertBlock(store.Block{ + Height: 401, + Hash: "other_network_hash", + RewardBT: 50, + FoundAt: time.Now(), + Source: "test", + Chain: "", // legacy — no chain recorded + }) + + // Canonical hash differs AND getblock fails (block doesn't exist on + // this network at all). Should NOT orphan — should stamp as "other". + rpc := mockRPC(t, func(method string) (any, *rpcErrResp) { + switch method { + case "getblockhash": + return "mainnet_canonical", nil + case "getblock": + return nil, &rpcErrResp{Code: -5, Message: "Block not found"} + } + return nil, nil + }) + + agg := newAgg(st, rpc) + setChain(agg, "main") + + agg.reconcileOnce(context.Background()) + + blocks, _ := st.Recent(10) + for _, b := range blocks { + if b.Height == 401 { + if !b.OrphanedAt.IsZero() { + t.Error("cross-network legacy block was falsely orphaned") + } + if b.Chain != "testnet4" { + t.Errorf("expected chain stamped as 'testnet4' (inferred from current=main), got %q", b.Chain) + } + } + } +} + // ---- IngestBlockEvents tests ---- // TestIngestBlockEvents_NewBlock verifies that a fresh block event is diff --git a/api/internal/store/blocks.go b/api/internal/store/blocks.go index 8057ef0..0e32df9 100644 --- a/api/internal/store/blocks.go +++ b/api/internal/store/blocks.go @@ -287,3 +287,41 @@ func (s *BlockStore) MarkOrphaned(height int64, at time.Time) error { ) return err } + +// UnmarkOrphaned clears the orphaned_at stamp set by MarkOrphaned. +// Used to correct false positives caused by a network switch: a block +// found on testnet is not "orphaned" when the node switches to mainnet — +// it simply belongs to a different chain and must not be compared against +// mainnet's canonical history. +func (s *BlockStore) UnmarkOrphaned(height int64) error { + _, err := s.db.Exec( + `UPDATE blocks SET orphaned_at = 0 WHERE height = ?`, + height, + ) + return err +} + +// UnmarkOrphanedStampChain clears orphaned_at and simultaneously sets the +// chain field. Used for legacy rows (chain == "") that were falsely orphaned +// during a network switch: by stamping a non-mainnet chain value we prevent +// the reconcile reorg-detection pass from re-checking the block against the +// current chain's canonical hashes. +func (s *BlockStore) UnmarkOrphanedStampChain(height int64, chain string) error { + _, err := s.db.Exec( + `UPDATE blocks SET orphaned_at = 0, chain = ? WHERE height = ?`, + chain, height, + ) + return err +} + +// StampChain sets the chain field on a block without touching orphaned_at. +// Used when the reconcile loop determines a block belongs to a different +// network — the block is not orphaned (a reorg), it just doesn't belong +// to the current chain. +func (s *BlockStore) StampChain(height int64, chain string) error { + _, err := s.db.Exec( + `UPDATE blocks SET chain = ? WHERE height = ?`, + chain, height, + ) + return err +}