Stop when the store is whole, not when a window is

v1.2.42 gave the heal a loop so a long history takes as many passes as it needs,
and then read the wrong number to decide it was done: the count of rows in the
current window it had FAILED to date. A window that heals completely has no
failures in it, so the very first success looked like completion. On the Lux
C-Chain that dated 50,000 of 912,492 rows and stopped, leaving 862,492 trades
still sorting to 1970 — better than before by one window, invisible by the same
mechanism.

A pass now reports what it found and what it dated. Nothing found is the only
way out; progress goes straight on to the next window; found-but-none-dated
waits and retries, which is the RPC being unreachable rather than the work being
over.

The test seeds more undated rows than one window holds and drains to zero.
Against v1.2.42 it stops with rows still holding a block number.

Co-authored-by: Hanzo Dev <[email protected]>

indexer/blocks.go

@@ -117,14 +117,14 @@ func (idx *Indexer) stampTimes(ctx context.Context, logs []logEntry) error {
 }
 
 // healSwapTimes gives stored swaps the real time of their block, and reports
-// how many it dated and how many still lack one.
+// how many rows it found needing one and how many it dated.
 //
 // A row written by an older build holds its block NUMBER in Timestamp and has
 // no Block at all. That is not a guess: no swap is mined in block 0, so an
 // unset Block identifies exactly those rows, and the number it left behind is
 // the very thing needed to look the time up. A row whose supposed block the
 // chain does not answer for is left alone — this heals what it can prove.
-func (idx *Indexer) healSwapTimes(ctx context.Context) (healed, remaining int, err error) {
+func (idx *Indexer) healSwapTimes(ctx context.Context) (found, healed int, err error) {
 	// Asked for by the property that identifies them. This read used to be
 	// RecentSwapsRaw — the newest rows by timestamp — which is the one window an
 	// undated row is never in: it holds a block number, block numbers are small
@@ -143,13 +143,14 @@ func (idx *Indexer) healSwapTimes(ctx context.Context) (healed, remaining int, e
 			blocks = append(blocks, b)
 		}
 	}
-	if len(stale) == 0 {
+	found = len(stale)
+	if found == 0 {
 		return 0, 0, nil
 	}
 
 	times := idx.blockTimes(ctx, blocks)
 	if len(times) == 0 {
-		return 0, len(stale), fmt.Errorf("no block header answered for %d blocks", len(blocks))
+		return found, 0, fmt.Errorf("no block header answered for %d blocks", len(blocks))
 	}
 	for id, sw := range stale {
 		block := uint64(sw.Timestamp)
@@ -162,8 +163,8 @@ func (idx *Indexer) healSwapTimes(ctx context.Context) (healed, remaining int, e
 		idx.store.SeedSwap(id, &next)
 		healed++
 	}
-	idx.logf("[indexer] swap times healed — %d of %d rows dated from their block header", healed, len(stale))
-	return healed, len(stale) - healed, nil
+	idx.logf("[indexer] swap times healed — %d of %d rows dated from their block header", healed, found)
+	return found, healed, nil
 }
 
 // keepSwapTimes runs the heal until no row is left holding a block number where
@@ -176,13 +177,19 @@ func (idx *Indexer) healSwapTimes(ctx context.Context) (healed, remaining int, e
 // past it undated, which is the same invisibility by a smaller margin.
 func (idx *Indexer) keepSwapTimes(ctx context.Context) {
 	for {
-		healed, stale, err := idx.healSwapTimes(ctx)
-		if stale == 0 && err == nil {
+		found, healed, err := idx.healSwapTimes(ctx)
+		// Nothing left holding a block number: the store is whole, and new
+		// swaps arrive already dated. Done — and this is the ONLY way out.
+		if found == 0 && err == nil {
 			return
 		}
 		if err != nil {
-			idx.logf("[indexer] swap times: %v (%d rows still undated)", err, stale)
+			idx.logf("[indexer] swap times: %v (%d rows found undated)", err, found)
 		}
+		// A pass covers one window. Progress means there is very likely another
+		// window behind it, so go straight on. Stopping on "this window has no
+		// failures left" is stopping when the window is DONE rather than when
+		// the store is — one window of 50,000 dated and 862,492 left behind.
 		if healed > 0 && err == nil {
 			continue
 		}

indexer/blocks_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/luxfi/graph/storage"
 )
@@ -186,9 +187,9 @@ func TestHealDatesTheHistory(t *testing.T) {
 	})
 	idx := NewWithConfig(Config{RPC: chainAt(t, mined, nil).URL}, s)
 
-	_, left, err := idx.healSwapTimes(context.Background())
-	if err != nil || left != 0 {
-		t.Fatalf("heal left %d rows undated: %v", left, err)
+	found, healed, err := idx.healSwapTimes(context.Background())
+	if err != nil || found != healed {
+		t.Fatalf("heal dated %d of %d rows found: %v", healed, found, err)
 	}
 	got := s.RecentSwapsRaw(10)
 	old := got["0xold#0x1"]
@@ -202,8 +203,8 @@ func TestHealDatesTheHistory(t *testing.T) {
 		t.Errorf("an already-dated row was rewritten: %+v", n)
 	}
 	// Idempotent: a second run has nothing to do and must not re-date anything.
-	if _, left, err := idx.healSwapTimes(context.Background()); left != 0 || err != nil {
-		t.Errorf("second heal reported %d undated rows: %v", left, err)
+	if found, _, err := idx.healSwapTimes(context.Background()); found != 0 || err != nil {
+		t.Errorf("second heal found %d undated rows: %v", found, err)
 	}
 }
 
@@ -232,9 +233,9 @@ func TestHealFindsRowsBuriedUnderDatedOnes(t *testing.T) {
 	}
 	idx := NewWithConfig(Config{RPC: chainAt(t, mined, nil).URL}, s)
 
-	healed, left, err := idx.healSwapTimes(context.Background())
-	if err != nil || left != 0 || healed != 1 {
-		t.Fatalf("heal dated %d rows and left %d undated (%v) — the buried row was not found", healed, left, err)
+	found, healed, err := idx.healSwapTimes(context.Background())
+	if err != nil || found != 1 || healed != 1 {
+		t.Fatalf("heal found %d and dated %d (%v) — the buried row was not found", found, healed, err)
 	}
 	got := s.UndatedSwapsRaw(10)
 	if len(got) != 0 {
@@ -263,3 +264,36 @@ func TestDayStartIsUTCMidnight(t *testing.T) {
 		t.Errorf("dayID = %q, want 0xabc-20454", got)
 	}
 }
+
+// A store holds more undated rows than one pass can window, and every one of
+// them has to end up dated.
+//
+// The heal reads a bounded window, so a long history takes several passes. The
+// loop that repeats it must stop when the STORE has nothing left, not when a
+// window does — a window that heals completely has no failures in it, and
+// reading that as "finished" stops after the first 50,000 and leaves the rest
+// exactly as invisible as before.
+func TestHealDrainsEveryWindow(t *testing.T) {
+	const mined = 1767225600 + 7200
+	s := newMemSQLiteStore(t)
+	const rows = maxValuedSwaps + 2500
+	for i := 0; i < rows; i++ {
+		s.SeedSwap(fmt.Sprintf("0xold#%d", i), &storage.SeedSwapData{
+			Timestamp: int64(1 + i%9000), // a block number, as an older build wrote it
+			Pool:      "0xpool", Amount0: "1", Amount1: "-2", AmountUSD: "0",
+		})
+	}
+	idx := NewWithConfig(Config{RPC: chainAt(t, mined, nil).URL}, s)
+
+	keepSwapTimesDone := make(chan struct{})
+	go func() { idx.keepSwapTimes(context.Background()); close(keepSwapTimesDone) }()
+	select {
+	case <-keepSwapTimesDone:
+	case <-time.After(90 * time.Second):
+		t.Fatal("heal never finished draining")
+	}
+
+	if left := len(s.UndatedSwapsRaw(10)); left != 0 {
+		t.Errorf("%d rows still hold a block number after the heal stopped", left)
+	}
+}