Add tests for the critical mining path

Tests cover the four subsystems that sit between a share submission and
a confirmed block reward:

  bitcoind/rpc_test.go  — isRetryable classification, 3-attempt backoff
    on 503, no retry on semantic RPC errors (-5), retry on warm-up (-28),
    CoinbaseReward calculation.

  logmon/tailer_test.go — handleLine regex parsing, share-diff carry from
    attempt line to block event, diff reset after use, unrelated-line
    silence, file-read via goroutine, cursor save/resume across restart,
    log rotation re-open.

  store/blocks_test.go  — InsertBlock roundtrip (all fields including
    chain), idempotent dedup, MarkOrphaned, UpdateEnrichment,
    BlocksNeedingEnrichment, KV get/set/overwrite, and migration of an
    existing DB that lacks the chain column.

  state/blocks_test.go  — reconcileOnce chain-filter (testnet block not
    orphaned on mainnet), genuine reorg detection (different canonical
    hash → orphaned), transient RPC error (no orphan on -5), legacy
    empty-chain block still checked for reorgs, IngestBlockEvents new
    block persisted + confirmed counter incremented, dedup skips counter
    and OnRefresh, chain stamped from current snapshot at ingest time.

api/internal/state/blocks_test.go

@@ -0,0 +1,378 @@
+package state
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/kamadopool/kamado-api/internal/bitcoind"
+	"github.com/kamadopool/kamado-api/internal/logmon"
+	"github.com/kamadopool/kamado-api/internal/store"
+)
+
+func discardLog() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func openTempStore(t *testing.T) *store.BlockStore {
+	t.Helper()
+	f, err := os.CreateTemp("", "state_test*.db")
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.Close()
+	t.Cleanup(func() { _ = os.Remove(f.Name()) })
+	s, err := store.Open(f.Name())
+	if err != nil {
+		t.Fatalf("store.Open: %v", err)
+	}
+	t.Cleanup(func() { _ = s.Close() })
+	return s
+}
+
+// mockRPC starts an httptest server whose responses are determined by the
+// supplied handler. The handler receives the RPC method name and the raw
+// params JSON and returns the value to place in "result". Returning nil
+// sends an empty result; returning a non-nil error (as rpcErrResp) sends
+// an RPC error response.
+type rpcErrResp struct {
+	Code    int
+	Message string
+}
+
+func mockRPC(t *testing.T, handler func(method string) (any, *rpcErrResp)) *bitcoind.RPC {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var req struct {
+			Method string `json:"method"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			http.Error(w, err.Error(), 400)
+			return
+		}
+		result, rpcErr := handler(req.Method)
+		if rpcErr != nil {
+			json.NewEncoder(w).Encode(map[string]any{
+				"result": nil,
+				"error":  map[string]any{"code": rpcErr.Code, "message": rpcErr.Message},
+			})
+			return
+		}
+		json.NewEncoder(w).Encode(map[string]any{"result": result, "error": nil})
+	}))
+	t.Cleanup(srv.Close)
+	return bitcoind.NewRPC(srv.URL, "u", "p", 5*time.Second)
+}
+
+func newAgg(st *store.BlockStore, rpc *bitcoind.RPC) *Aggregator {
+	return &Aggregator{
+		Store: st,
+		RPC:   rpc,
+		Log:   discardLog(),
+		ready: make(chan struct{}),
+	}
+}
+
+// setChain sets the aggregator's snapshot chain without going through refresh.
+func setChain(agg *Aggregator, chain string) {
+	agg.mu.Lock()
+	agg.snap.Chain = &bitcoind.BlockchainInfo{Chain: chain}
+	agg.mu.Unlock()
+}
+
+// ---- reconcileOnce tests ----
+
+// TestReconcileOnce_ChainFilter verifies that a block recorded on testnet
+// is NOT orphaned when the node is now on mainnet — even if the canonical
+// hash at that height on mainnet is different.
+func TestReconcileOnce_ChainFilter_NoFalseOrphan(t *testing.T) {
+	st := openTempStore(t)
+
+	// Insert a testnet block with a known hash.
+	_, err := st.InsertBlock(store.Block{
+		Height:   100,
+		Hash:     "testnet_hash_aaa",
+		RewardBT: 50,
+		FoundAt:  time.Now(),
+		Source:   "test",
+		Chain:    "test",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Mock RPC returns a DIFFERENT hash for height 100 — mainnet canonical.
+	rpc := mockRPC(t, func(method string) (any, *rpcErrResp) {
+		if method == "getblockhash" {
+			return "mainnet_hash_bbb", nil
+		}
+		return nil, nil
+	})
+
+	agg := newAgg(st, rpc)
+	setChain(agg, "main") // switched to mainnet
+
+	agg.reconcileOnce(context.Background())
+
+	blocks, err := st.Recent(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, b := range blocks {
+		if b.Height == 100 && !b.OrphanedAt.IsZero() {
+			t.Error("testnet block was falsely orphaned after switching to mainnet — chain filter broken")
+		}
+	}
+}
+
+// TestReconcileOnce_GenuineReorg verifies that when the canonical block at a
+// recorded height has a different hash (real on-chain reorg), the block is
+// marked orphaned.
+func TestReconcileOnce_GenuineReorg(t *testing.T) {
+	st := openTempStore(t)
+
+	_, err := st.InsertBlock(store.Block{
+		Height:   200,
+		Hash:     "our_hash_aaaa",
+		RewardBT: 3.125,
+		FoundAt:  time.Now(),
+		Source:   "test",
+		Chain:    "main",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rpc := mockRPC(t, func(method string) (any, *rpcErrResp) {
+		if method == "getblockhash" {
+			return "canonical_bbbb", nil // different — reorg happened
+		}
+		return nil, nil
+	})
+
+	agg := newAgg(st, rpc)
+	setChain(agg, "main")
+
+	agg.reconcileOnce(context.Background())
+
+	blocks, err := st.Recent(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := false
+	for _, b := range blocks {
+		if b.Height == 200 {
+			found = true
+			if b.OrphanedAt.IsZero() {
+				t.Error("block at reorged height was not marked orphaned")
+			}
+		}
+	}
+	if !found {
+		t.Error("block 200 not found in store after reconcile")
+	}
+}
+
+// TestReconcileOnce_RPCError_NoOrphan verifies that a transient RPC error
+// (getblockhash fails) does not cause a block to be orphaned. The next
+// reconcile sweep will retry.
+func TestReconcileOnce_RPCError_NoOrphan(t *testing.T) {
+	st := openTempStore(t)
+
+	st.InsertBlock(store.Block{
+		Height:   300,
+		Hash:     "our_hash_cccc",
+		RewardBT: 3.125,
+		FoundAt:  time.Now(),
+		Source:   "test",
+		Chain:    "main",
+	})
+
+	rpc := mockRPC(t, func(method string) (any, *rpcErrResp) {
+		if method == "getblockhash" {
+			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 == 300 && !b.OrphanedAt.IsZero() {
+			t.Error("block was orphaned on RPC error — should only orphan on definitive hash mismatch")
+		}
+	}
+}
+
+// TestReconcileOnce_LegacyBlock_StillChecked 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) {
+	st := openTempStore(t)
+
+	st.InsertBlock(store.Block{
+		Height:   400,
+		Hash:     "legacy_hash",
+		RewardBT: 3.125,
+		FoundAt:  time.Now(),
+		Source:   "test",
+		Chain:    "", // legacy — no chain recorded
+	})
+
+	// Mainnet canonical hash differs: should orphan even though Chain is empty.
+	rpc := mockRPC(t, func(method string) (any, *rpcErrResp) {
+		if method == "getblockhash" {
+			return "different_canonical", nil
+		}
+		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 == 400 && b.OrphanedAt.IsZero() {
+			t.Error("legacy block (empty chain) should still be checked for reorgs")
+		}
+	}
+}
+
+// ---- IngestBlockEvents tests ----
+
+// TestIngestBlockEvents_NewBlock verifies that a fresh block event is
+// persisted to the store and the confirmed counter increments.
+func TestIngestBlockEvents_NewBlock(t *testing.T) {
+	st := openTempStore(t)
+	agg := newAgg(st, nil) // nil RPC skips enrichment
+	setChain(agg, "main")
+
+	events := make(chan logmon.BlockEvent, 1)
+
+	done := make(chan struct{})
+	agg.OnRefresh = func(_ Snapshot) { close(done) }
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	go agg.IngestBlockEvents(ctx, events)
+
+	events <- logmon.BlockEvent{Height: 500, SeenAt: time.Now(), ShareDiff: 32768}
+
+	select {
+	case <-done:
+	case <-ctx.Done():
+		t.Fatal("timed out waiting for block to be processed")
+	}
+
+	blocks, _ := st.Recent(10)
+	if len(blocks) == 0 {
+		t.Fatal("block not persisted to store")
+	}
+	if blocks[0].Height != 500 {
+		t.Errorf("Height = %d, want 500", blocks[0].Height)
+	}
+	if blocks[0].Chain != "main" {
+		t.Errorf("Chain = %q, want main", blocks[0].Chain)
+	}
+
+	agg.mu.RLock()
+	confirmed := agg.blockSubmitsConfirmed
+	agg.mu.RUnlock()
+	if confirmed != 1 {
+		t.Errorf("blockSubmitsConfirmed = %d, want 1", confirmed)
+	}
+}
+
+// TestIngestBlockEvents_Dedup verifies that sending the same block height
+// twice does not double-count the confirmed counter or call OnRefresh twice.
+func TestIngestBlockEvents_Dedup(t *testing.T) {
+	st := openTempStore(t)
+
+	// Pre-insert the block so the first ingest sees it as a duplicate.
+	st.InsertBlock(store.Block{Height: 600, FoundAt: time.Now(), Chain: "main"})
+
+	var refreshCount int
+	var mu sync.Mutex
+
+	agg := newAgg(st, nil)
+	agg.OnRefresh = func(_ Snapshot) {
+		mu.Lock()
+		refreshCount++
+		mu.Unlock()
+	}
+	setChain(agg, "main")
+
+	events := make(chan logmon.BlockEvent, 1)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		agg.IngestBlockEvents(ctx, events)
+	}()
+
+	events <- logmon.BlockEvent{Height: 600, SeenAt: time.Now()}
+	close(events)
+	wg.Wait()
+
+	agg.mu.RLock()
+	confirmed := agg.blockSubmitsConfirmed
+	agg.mu.RUnlock()
+	if confirmed != 0 {
+		t.Errorf("blockSubmitsConfirmed = %d, want 0 for duplicate", confirmed)
+	}
+
+	mu.Lock()
+	rc := refreshCount
+	mu.Unlock()
+	if rc != 0 {
+		t.Errorf("OnRefresh called %d times, want 0 for duplicate block", rc)
+	}
+}
+
+// TestIngestBlockEvents_ChainStamping verifies that the block record stored
+// carries the chain name from the current snapshot at ingest time.
+func TestIngestBlockEvents_ChainStamping(t *testing.T) {
+	st := openTempStore(t)
+	agg := newAgg(st, nil)
+	setChain(agg, "test") // testnet at the time of the solve
+
+	events := make(chan logmon.BlockEvent, 1)
+	done := make(chan struct{})
+	agg.OnRefresh = func(_ Snapshot) { close(done) }
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	go agg.IngestBlockEvents(ctx, events)
+
+	events <- logmon.BlockEvent{Height: 700, SeenAt: time.Now()}
+
+	select {
+	case <-done:
+	case <-ctx.Done():
+		t.Fatal("timed out waiting for block to be processed")
+	}
+
+	blocks, _ := st.Recent(10)
+	if len(blocks) == 0 {
+		t.Fatal("block not found in store")
+	}
+	if blocks[0].Chain != "test" {
+		t.Errorf("Chain = %q, want test — chain stamping broken", blocks[0].Chain)
+	}
+}