fix(server): back off, retry, and de-race the EVM indexer fetch loop

This PR set out to adopt cosmos/evm's EVMIndexerService for its
wait-before-retry on Block/BlockResults fetch errors; review (phenix) caught
that upstream's version (still on their main) latches the fetch error and
never clears it — after the first transient failure the loop only sleeps and
indexing stalls until restart. The old in-tree copy had the opposite defect:
it busy-looped the failing fetch with no backoff. Writing the regression test
also surfaced a third defect present in both: the latest-height variable is
written by the new-block subscription goroutine and read by the indexing loop
unsynchronized (go test -race fails on both prior versions).

Keep the service in-tree on upstream's structure with all three fixed:
back off on fetch errors (upstream's improvement), clear the error before
retrying (the missing line), and make latestBlock an atomic.

The regression test drives the loop with a scripted RPC client and fails on
each historical defect: it times out on the latched-stall version, exceeds
the bounded fetch-attempt count on the busy-loop version, and trips -race on
the unsynchronized version.

Signed-off-by: puneetmahajan <59960662+puneet2019@users.noreply.github.com>

Author: puneet2019

CHANGELOG.md

@@ -70,6 +70,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 ### Bug Fixes
 
+- (server) [#311](https://github.com/mocachain/moca/pull/311) Harden the EVM tx indexer service's fetch loop: back off on transient `Block`/`BlockResults` fetch errors instead of busy-looping, clear the error before retrying so indexing cannot stall until restart (upstream cosmos/evm as of v0.6.0 latches it), and make the shared latest-height access atomic (data race between the new-block subscription goroutine and the indexing loop). Guarded by a mock-driven regression test that fails on all three defects.
 - (virtualgroup,storage) [#306](https://github.com/mocachain/moca/pull/306) Tighten storage provider exit preconditions and make discontinued-resource cleanup resolve its primary SP defensively.
 - (storage) [#298](https://github.com/mocachain/moca/pull/298) Close the object iterator in `isNonEmptyBucket` to fix a per-call store-iterator leak (MOCA-413)
 - (sp) [#299](https://github.com/mocachain/moca/pull/299) Close the iterator in `GetAllStorageProviders` to fix a store-iterator leak

server/indexer_service.go

@@ -2,11 +2,14 @@ package server
 
 import (
 	"context"
+	"sync/atomic"
 	"time"
 
 	"github.com/cometbft/cometbft/libs/service"
 	rpcclient "github.com/cometbft/cometbft/rpc/client"
+	coretypes "github.com/cometbft/cometbft/rpc/core/types"
 	"github.com/cometbft/cometbft/types"
+
 	servertypes "github.com/cosmos/evm/server/types"
 )
 
@@ -42,7 +45,11 @@ func (eis *EVMIndexerService) OnStart() error {
 	if err != nil {
 		return err
 	}
-	latestBlock := status.SyncInfo.LatestBlockHeight
+	// latestBlock is written by the new-block subscription goroutine and read
+	// by the indexing loop below — keep the access atomic (data race in the
+	// pre-fix copy and in upstream cosmos/evm as of v0.6.0).
+	var latestBlock atomic.Int64
+	latestBlock.Store(status.SyncInfo.LatestBlockHeight)
 	newBlockSignal := make(chan struct{}, 1)
 
 	// Use SubscribeUnbuffered here to ensure both subscriptions does not get
@@ -61,8 +68,8 @@ func (eis *EVMIndexerService) OnStart() error {
 		for {
 			msg := <-blockHeadersChan
 			eventDataHeader := msg.Data.(types.EventDataNewBlockHeader)
-			if eventDataHeader.Header.Height > latestBlock {
-				latestBlock = eventDataHeader.Header.Height
+			if eventDataHeader.Header.Height > latestBlock.Load() {
+				latestBlock.Store(eventDataHeader.Header.Height)
 				// notify
 				select {
 				case newBlockSignal <- struct{}{}:
@@ -77,26 +84,45 @@ func (eis *EVMIndexerService) OnStart() error {
 		return err
 	}
 	if lastBlock == -1 {
-		lastBlock = latestBlock
+		lastBlock = latestBlock.Load()
 	}
+
+	// blockErr indicates an error fetching an expected block or its results
+	var blockErr error
+
 	for {
-		if latestBlock <= lastBlock {
-			// nothing to index. wait for signal of new block
+		if latestBlock.Load() <= lastBlock || blockErr != nil {
+			// two cases to enter this block:
+			// 1. nothing to index (indexer is caught up). wait for signal of new block.
+			// 2. previous attempt to index errored (failed to fetch the Block or BlockResults).
+			//    in this case, wait before retrying the data fetching, rather than infinite looping
+			//    a failing fetch. this can occur due to drive latency between the block existing and its
+			//    block_results getting saved.
 			select {
 			case <-newBlockSignal:
 			case <-time.After(NewBlockWaitTimeout):
 			}
+			// Clear the latched fetch error so the loop actually re-attempts
+			// the fetch after backing off. Without this (upstream cosmos/evm
+			// as of v0.6.0), the first transient Block/BlockResults failure
+			// gates the loop forever and indexing stalls until restart.
+			blockErr = nil
 			continue
 		}
-		for i := lastBlock + 1; i <= latestBlock; i++ {
-			block, err := eis.client.Block(ctx, &i)
-			if err != nil {
-				eis.Logger.Error("failed to fetch block", "height", i, "err", err)
+		for i := lastBlock + 1; i <= latestBlock.Load(); i++ {
+			var (
+				block       *coretypes.ResultBlock
+				blockResult *coretypes.ResultBlockResults
+			)
+
+			block, blockErr = eis.client.Block(ctx, &i)
+			if blockErr != nil {
+				eis.Logger.Error("failed to fetch block", "height", i, "err", blockErr)
 				break
 			}
-			blockResult, err := eis.client.BlockResults(ctx, &i)
-			if err != nil {
-				eis.Logger.Error("failed to fetch block result", "height", i, "err", err)
+			blockResult, blockErr = eis.client.BlockResults(ctx, &i)
+			if blockErr != nil {
+				eis.Logger.Error("failed to fetch block result", "height", i, "err", blockErr)
 				break
 			}
 			if err := eis.txIdxr.IndexBlock(block.Block, blockResult.TxsResults); err != nil {

server/indexer_service_test.go

@@ -0,0 +1,130 @@
+package server
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	abci "github.com/cometbft/cometbft/abci/types"
+	rpcclient "github.com/cometbft/cometbft/rpc/client"
+	coretypes "github.com/cometbft/cometbft/rpc/core/types"
+	cmttypes "github.com/cometbft/cometbft/types"
+	servertypes "github.com/cosmos/evm/server/types"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+// mockRPCClient scripts the only four rpcclient.Client methods the indexer
+// service uses. BlockResults fails until failuresLeft drains, tracking the
+// attempt count — the seam for both indexer-loop regressions.
+type mockRPCClient struct {
+	rpcclient.Client // nil-embedded: any unscripted call panics loudly
+
+	height       int64
+	headerCh     chan coretypes.ResultEvent
+	failuresLeft atomic.Int64
+	attempts     atomic.Int64
+}
+
+func (m *mockRPCClient) Status(context.Context) (*coretypes.ResultStatus, error) {
+	return &coretypes.ResultStatus{
+		SyncInfo: coretypes.SyncInfo{LatestBlockHeight: m.height},
+	}, nil
+}
+
+func (m *mockRPCClient) Subscribe(_ context.Context, _, _ string, _ ...int) (<-chan coretypes.ResultEvent, error) {
+	return m.headerCh, nil
+}
+
+func (m *mockRPCClient) Block(_ context.Context, height *int64) (*coretypes.ResultBlock, error) {
+	return &coretypes.ResultBlock{
+		Block: &cmttypes.Block{Header: cmttypes.Header{Height: *height}},
+	}, nil
+}
+
+func (m *mockRPCClient) BlockResults(_ context.Context, height *int64) (*coretypes.ResultBlockResults, error) {
+	m.attempts.Add(1)
+	if m.failuresLeft.Add(-1) >= 0 {
+		return nil, context.DeadlineExceeded
+	}
+	return &coretypes.ResultBlockResults{Height: *height}, nil
+}
+
+// mockIndexer records indexed heights; the query methods are never used by
+// the service loop.
+type mockIndexer struct {
+	indexed chan int64
+}
+
+func (m *mockIndexer) LastIndexedBlock() (int64, error)  { return 0, nil }
+func (m *mockIndexer) FirstIndexedBlock() (int64, error) { return 0, nil }
+func (m *mockIndexer) IndexBlock(block *cmttypes.Block, _ []*abci.ExecTxResult) error {
+	m.indexed <- block.Height
+	return nil
+}
+
+func (m *mockIndexer) GetByTxHash(common.Hash) (*servertypes.TxResult, error) {
+	panic("not used")
+}
+
+func (m *mockIndexer) GetByBlockAndIndex(int64, int32) (*servertypes.TxResult, error) {
+	panic("not used")
+}
+
+// TestIndexerServiceRetriesAfterFetchError guards both historical failure
+// modes of the indexing loop around transient Block/BlockResults errors:
+//
+//   - stall (upstream cosmos/evm as of v0.6.0): the fetch error is latched and
+//     never cleared, so after the first failure the loop only sleeps and the
+//     indexer never progresses again — this test times out on that code;
+//   - busy-loop (moca's pre-fix copy): the loop hammered the failing fetch
+//     with no backoff — the bounded attempt count below fails on that code.
+//
+// Correct behavior: back off on error, then re-attempt on the next new-block
+// signal (or timeout) and recover.
+func TestIndexerServiceRetriesAfterFetchError(t *testing.T) {
+	client := &mockRPCClient{
+		height:   1,
+		headerCh: make(chan coretypes.ResultEvent, 4),
+	}
+	client.failuresLeft.Store(2) // first two BlockResults calls fail, then succeed
+	idxr := &mockIndexer{indexed: make(chan int64, 8)}
+
+	svc := NewEVMIndexerService(idxr, client)
+	go func() {
+		_ = svc.OnStart() // loops forever; test asserts via channels
+	}()
+
+	signal := func(height int64) {
+		client.headerCh <- coretypes.ResultEvent{
+			Data: cmttypes.EventDataNewBlockHeader{
+				Header: cmttypes.Header{Height: height},
+			},
+		}
+	}
+
+	// Attempt 1 fails on startup catch-up (height 1). Each signal wakes the
+	// wait branch: attempt 2 fails, attempt 3 succeeds and must recover.
+	signal(2)
+	signal(3)
+
+	deadline := time.After(10 * time.Second)
+	got := map[int64]bool{}
+	for len(got) < 2 {
+		select {
+		case h := <-idxr.indexed:
+			got[h] = true
+		case <-deadline:
+			t.Fatalf("indexer stalled after transient fetch error; indexed so far: %v (attempts=%d)",
+				got, client.attempts.Load())
+		}
+	}
+	require.True(t, got[1] && got[2], "heights 1 and 2 must be indexed after recovery, got %v", got)
+
+	// No busy-loop: while erroring, exactly one fetch attempt per wake-up.
+	// Catch-up + two signaled retries + the follow-up block ≈ 4 attempts;
+	// leave slack for scheduling, but orders of magnitude below a busy loop.
+	require.LessOrEqual(t, client.attempts.Load(), int64(6),
+		"fetch attempts should be bounded by wake-ups (busy-loop regression)")
+}