From 445bcad0aa34fcf132a929aa7c54933bc7e9dcf9 Mon Sep 17 00:00:00 2001 From: vietddude Date: Thu, 18 Jun 2026 17:12:05 +0700 Subject: [PATCH 1/2] fix: treat TLS errors as failover-worthy and harden start-block fallback --- internal/rpc/failover.go | 6 ++ internal/rpc/failover_test.go | 12 +++ internal/worker/regular.go | 141 ++++++++++++++--------------- internal/worker/regular_test.go | 76 +++++++++++++++- pkg/store/blockstore/store.go | 6 ++ pkg/store/blockstore/store_test.go | 55 +++++++++++ 6 files changed, 221 insertions(+), 75 deletions(-) create mode 100644 pkg/store/blockstore/store_test.go diff --git a/internal/rpc/failover.go b/internal/rpc/failover.go index ec63173..2904c72 100644 --- a/internal/rpc/failover.go +++ b/internal/rpc/failover.go @@ -666,6 +666,12 @@ func (f *Failover[T]) analyzeError(err error, elapsed time.Duration) ProviderIss cooldown: 2 * time.Minute, markUnhealthy: true, }, + { + patterns: []string{"tls: internal error", "tls handshake", "handshake failure", "remote error: tls"}, + reason: "tls_error", + cooldown: 2 * time.Minute, + markUnhealthy: true, + }, { patterns: []string{"-32007", "batch limit exceeded", "internal error -32005"}, reason: "batch_limit", diff --git a/internal/rpc/failover_test.go b/internal/rpc/failover_test.go index b1a7009..0774374 100644 --- a/internal/rpc/failover_test.go +++ b/internal/rpc/failover_test.go @@ -130,6 +130,18 @@ func TestAnalyzeAndHandleError_ConnectionError(t *testing.T) { assert.Equal(t, int64(1), errorsByType["connection_error"]) } +func TestAnalyzeAndHandleError_TLSError(t *testing.T) { + f, p := newTestFailover() + + err := fmt.Errorf("eth_blockNumber failed: remote error: tls: internal error") + f.AnalyzeAndHandleError(p, err, 100*time.Millisecond) + + assert.False(t, p.IsAvailable(), "provider should be blacklisted after TLS error") + + errorsByType := f.GetMetrics()["errors_by_type"].(map[string]int64) + assert.Equal(t, int64(1), errorsByType["tls_error"]) +} + func TestAnalyzeAndHandleError_GenericError(t *testing.T) { f, p := newTestFailover() diff --git a/internal/worker/regular.go b/internal/worker/regular.go index 4d35d76..e9a281d 100644 --- a/internal/worker/regular.go +++ b/internal/worker/regular.go @@ -13,6 +13,7 @@ import ( "github.com/fystack/multichain-indexer/pkg/common/enum" "github.com/fystack/multichain-indexer/pkg/events" "github.com/fystack/multichain-indexer/pkg/infra" + "github.com/fystack/multichain-indexer/pkg/retry" "github.com/fystack/multichain-indexer/pkg/store/blockstore" "github.com/fystack/multichain-indexer/pkg/store/pubkeystore" ) @@ -21,6 +22,8 @@ const ( MaxBlockHashSize = 50 regularGapRetryAttempts = 2 regularGapRetryDelay = time.Second + startBlockRetryAttempts = 3 + startBlockRetryDelay = 2 * time.Second ) var errRegularRecoveryReorgHandled = errors.New("regular recovery reorg handled") @@ -33,6 +36,8 @@ type RegularWorker struct { blockHashes []blockstore.BlockHashEntry hashesModified bool persistTicker *time.Ticker + // Zero uses the startBlockRetryDelay default; overridable in tests. + startBlockRetryDelay time.Duration } func NewRegularWorker( @@ -181,64 +186,86 @@ func (rw *RegularWorker) processRegularBlocks() error { return processErr } +// determineStartingBlock picks a start block from the chain head and the last +// indexed block. GetLatestBlock returns (0, nil) for a cold start, so a non-nil +// kvErr means the store is genuinely down: we must never rewind to config.StartBlock. func (rw *RegularWorker) determineStartingBlock() uint64 { - registry := status.EnsureStatusRegistry(rw.statusRegistry) - chainLatest, err1 := rw.chain.GetLatestBlockNumber(rw.ctx) - kvLatest, err2 := rw.blockStore.GetLatestBlock(rw.chain.GetNetworkInternalCode()) + chainLatest, chainErr := rw.getLatestBlockWithRetry() + kvLatest, kvErr := rw.blockStore.GetLatestBlock(rw.chain.GetNetworkInternalCode()) - if err1 != nil && err2 != nil { - rw.logger.Warn("Cannot get latest block from chain or KV, using config.StartBlock", - "chain", rw.chain.GetName(), - "startBlock", rw.config.StartBlock, - ) + switch { + case kvErr != nil && chainErr == nil: + rw.logger.Error("Block store unavailable, starting from chain head to avoid stale rewind", + "chain", rw.chain.GetName(), "chainLatest", chainLatest, "error", kvErr) + return chainLatest + + case kvErr != nil: + rw.logger.Error("Chain RPC and block store both unreachable after retries; using config.StartBlock as last resort", + "chain", rw.chain.GetName(), "startBlock", rw.config.StartBlock, "error", kvErr) return uint64(rw.config.StartBlock) - } - if err1 != nil && kvLatest > 0 { + case kvLatest == 0 && chainErr == nil: // cold start, chain reachable + return chainLatest + + case kvLatest == 0: + rw.logger.Warn("Cold start and chain RPC unavailable, using config.StartBlock", + "chain", rw.chain.GetName(), "startBlock", rw.config.StartBlock) + return uint64(rw.config.StartBlock) + + case chainErr != nil: rw.logger.Warn("Chain RPC failed, resuming from KV latest", - "chain", rw.chain.GetName(), - "kvLatest", kvLatest, - ) + "chain", rw.chain.GetName(), "kvLatest", kvLatest) return kvLatest } - if err2 != nil || kvLatest == 0 { - return chainLatest - } - + // Both available: resume from KV, queuing any gap up to chain head. if chainLatest > kvLatest { - start := kvLatest + 1 - end := chainLatest - - // Split the range into manageable chunks - ranges := splitCatchupRange(blockstore.CatchupRange{ - Start: start, End: end, Current: start - 1, - }, MAX_RANGE_SIZE) - - // Batch save all split ranges - if err := rw.blockStore.SaveCatchupRanges( - rw.chain.GetNetworkInternalCode(), - ranges, - ); err != nil { - rw.logger.Error("Failed to batch save catchup ranges", - "chain", rw.chain.GetName(), - "count", len(ranges), - "error", err, - ) - } else { - registry.UpsertCatchupRanges(rw.chain.GetName(), ranges) - } - + ranges := rw.queueCatchupRanges(kvLatest+1, chainLatest) rw.logger.Info("Queued catchup ranges", "chain", rw.chain.GetName(), - "gap", fmt.Sprintf("%d-%d", start, end), + "gap", fmt.Sprintf("%d-%d", kvLatest+1, chainLatest), "ranges_created", len(ranges), ) } - return chainLatest } +// queueCatchupRanges splits [start, end] into chunks, persists them, and +// reflects them in the status registry, returning the ranges produced. +func (rw *RegularWorker) queueCatchupRanges(start, end uint64) []blockstore.CatchupRange { + ranges := splitCatchupRange(blockstore.CatchupRange{ + Start: start, End: end, Current: start - 1, + }, MAX_RANGE_SIZE) + + if err := rw.blockStore.SaveCatchupRanges(rw.chain.GetNetworkInternalCode(), ranges); err != nil { + rw.logger.Error("Failed to save catchup ranges", + "chain", rw.chain.GetName(), + "count", len(ranges), + "error", err, + ) + return ranges + } + + status.EnsureStatusRegistry(rw.statusRegistry).UpsertCatchupRanges(rw.chain.GetName(), ranges) + return ranges +} + +// getLatestBlockWithRetry fetches the chain head, retrying transient RPC +// failures so a momentary provider hiccup at startup doesn't force a fallback. +func (rw *RegularWorker) getLatestBlockWithRetry() (uint64, error) { + delay := rw.startBlockRetryDelay + if delay <= 0 { + delay = startBlockRetryDelay + } + var latest uint64 + err := retry.Constant(func() error { + var err error + latest, err = rw.chain.GetLatestBlockNumber(rw.ctx) + return err + }, delay, startBlockRetryAttempts) + return latest, err +} + func (rw *RegularWorker) detectAndHandleReorg(res *indexer.BlockResult) (bool, error) { prevNum := res.Block.Number - 1 storedHash := rw.getBlockHash(prevNum) @@ -354,7 +381,6 @@ func (rw *RegularWorker) flushBlockHashes() { // skipAheadIfLagging checks if the regular worker is too far behind the chain head. // If so, it queues the skipped range for catchup and jumps currentBlock to chain head. func (rw *RegularWorker) skipAheadIfLagging(latest uint64) bool { - registry := status.EnsureStatusRegistry(rw.statusRegistry) maxLag := rw.config.MaxLag if maxLag == 0 { maxLag = constant.DefaultMaxLag @@ -376,22 +402,7 @@ func (rw *RegularWorker) skipAheadIfLagging(latest uint64) bool { "catchup_range", fmt.Sprintf("%d-%d", skipStart, skipEnd), ) - ranges := splitCatchupRange(blockstore.CatchupRange{ - Start: skipStart, End: skipEnd, Current: skipStart - 1, - }, MAX_RANGE_SIZE) - - if err := rw.blockStore.SaveCatchupRanges( - rw.chain.GetNetworkInternalCode(), - ranges, - ); err != nil { - rw.logger.Error("Failed to save skip-ahead catchup ranges", - "chain", rw.chain.GetName(), - "count", len(ranges), - "error", err, - ) - } else { - registry.UpsertCatchupRanges(rw.chain.GetName(), ranges) - } + ranges := rw.queueCatchupRanges(skipStart, skipEnd) rw.currentBlock = latest _ = rw.blockStore.SaveLatestBlock(rw.chain.GetNetworkInternalCode(), latest-1) @@ -593,20 +604,6 @@ func checkContinuity(prev, curr indexer.BlockResult) bool { return prev.Block.Hash == curr.Block.ParentHash } -func blockResultNumber(res indexer.BlockResult) uint64 { - if res.Block != nil { - return res.Block.Number - } - return res.Number -} - -func blockResultHash(res indexer.BlockResult) string { - if res.Block != nil { - return res.Block.Hash - } - return "" -} - func blockResultError(res indexer.BlockResult) string { if res.Error != nil { return res.Error.Message diff --git a/internal/worker/regular_test.go b/internal/worker/regular_test.go index 73573f1..d91a6b8 100644 --- a/internal/worker/regular_test.go +++ b/internal/worker/regular_test.go @@ -173,6 +173,73 @@ func TestRegularWorkerDetermineStartingBlockUpdatesCatchupRegistry(t *testing.T) require.Equal(t, uint64(5), resp.Networks[0].CatchupPendingBlocks) } +func TestRegularWorkerDetermineStartingBlockStoreDownChainUp(t *testing.T) { + t.Parallel() + + // Store is genuinely down but the chain is reachable: must start from chain + // head, never rewind to config.StartBlock. + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 500} + store := &stubBlockStore{getLatestBlockErr: errors.New("redis down")} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.StartBlock = 100 + + require.Equal(t, uint64(500), rw.determineStartingBlock()) +} + +func TestRegularWorkerDetermineStartingBlockStoreAndChainDown(t *testing.T) { + t.Parallel() + + // Both store and chain are down after retries: config.StartBlock is the + // explicit last resort. + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latestErr: errors.New("tls: internal error")} + store := &stubBlockStore{getLatestBlockErr: errors.New("redis down")} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.StartBlock = 100 + rw.startBlockRetryDelay = time.Millisecond + + require.Equal(t, uint64(100), rw.determineStartingBlock()) +} + +func TestRegularWorkerDetermineStartingBlockColdStartChainUp(t *testing.T) { + t.Parallel() + + // Store reachable with no prior block (cold start) and chain reachable: + // start from chain head, not config.StartBlock. + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 500} + store := &stubBlockStore{latestBlock: 0} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.StartBlock = 100 + + require.Equal(t, uint64(500), rw.determineStartingBlock()) +} + +func TestRegularWorkerDetermineStartingBlockColdStartChainDown(t *testing.T) { + t.Parallel() + + // Genuine cold start (no prior block) and chain unavailable: fall back to + // config.StartBlock. + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latestErr: errors.New("tls: internal error")} + store := &stubBlockStore{latestBlock: 0} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.StartBlock = 100 + rw.startBlockRetryDelay = time.Millisecond + + require.Equal(t, uint64(100), rw.determineStartingBlock()) +} + +func TestRegularWorkerDetermineStartingBlockChainDownResumesFromKV(t *testing.T) { + t.Parallel() + + // Chain RPC down but store has a known prior block: resume from KV latest. + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latestErr: errors.New("tls: internal error")} + store := &stubBlockStore{latestBlock: 42} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.StartBlock = 100 + rw.startBlockRetryDelay = time.Millisecond + + require.Equal(t, uint64(42), rw.determineStartingBlock()) +} + func TestCatchupWorkerUpdatesCatchupRegistryOnProgressAndCompletion(t *testing.T) { t.Parallel() @@ -253,6 +320,7 @@ type stubIndexer struct { internalCode string networkType enum.NetworkType latest uint64 + latestErr error getBlocksFunc func(ctx context.Context, from, to uint64, isParallel bool) ([]indexer.BlockResult, error) getBlockFunc func(ctx context.Context, number uint64) (*types.Block, error) getBlockCalls []uint64 @@ -271,6 +339,9 @@ func (s *stubIndexer) GetNetworkInternalCode() string { } func (s *stubIndexer) GetLatestBlockNumber(context.Context) (uint64, error) { + if s.latestErr != nil { + return 0, s.latestErr + } return s.latest, nil } @@ -315,9 +386,8 @@ func (s *stubBlockStore) GetLatestBlock(string) (uint64, error) { if s.getLatestBlockErr != nil { return 0, s.getLatestBlockErr } - if s.latestBlock == 0 { - return 0, errors.New("not found") - } + // Mirrors the real store contract: a missing key (cold start) returns + // (0, nil); only genuine store errors return a non-nil error. return s.latestBlock, nil } diff --git a/pkg/store/blockstore/store.go b/pkg/store/blockstore/store.go index 69c2729..30019eb 100644 --- a/pkg/store/blockstore/store.go +++ b/pkg/store/blockstore/store.go @@ -11,6 +11,7 @@ import ( "github.com/fystack/multichain-indexer/pkg/common/constant" "github.com/fystack/multichain-indexer/pkg/common/logger" "github.com/fystack/multichain-indexer/pkg/infra" + "github.com/fystack/multichain-indexer/pkg/kvstore" ) // BlockHashEntry represents a block number and its hash for reorg detection. @@ -100,6 +101,11 @@ func NewBlockStore(store infra.KVStore) Store { func (bs *blockStore) GetLatestBlock(chainName string) (uint64, error) { startBlock, err := bs.store.Get(latestBlockKey(chainName)) if err != nil { + // Missing key (cold start) returns (0, nil); only real store errors + // propagate, so callers never silently rewind to config.StartBlock. + if errors.Is(err, kvstore.ErrKeyNotFound) { + return 0, nil + } return 0, err } return strconv.ParseUint(startBlock, 10, 64) diff --git a/pkg/store/blockstore/store_test.go b/pkg/store/blockstore/store_test.go new file mode 100644 index 0000000..142ecfa --- /dev/null +++ b/pkg/store/blockstore/store_test.go @@ -0,0 +1,55 @@ +package blockstore + +import ( + "errors" + "testing" + + "github.com/fystack/multichain-indexer/pkg/infra" + "github.com/fystack/multichain-indexer/pkg/kvstore" + "github.com/hashicorp/consul/api" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeKVStore is a minimal infra.KVStore whose Get behavior is configurable. +type fakeKVStore struct { + getVal string + getErr error +} + +func (f *fakeKVStore) GetName() string { return "fake" } +func (f *fakeKVStore) Set(string, string) error { return nil } +func (f *fakeKVStore) Get(string) (string, error) { return f.getVal, f.getErr } +func (f *fakeKVStore) SetAny(string, any) error { return nil } +func (f *fakeKVStore) GetAny(string, any) (bool, error) { return false, nil } +func (f *fakeKVStore) List(string) ([]*infra.KVPair, error) { + return nil, nil +} +func (f *fakeKVStore) Delete(string) error { return nil } +func (f *fakeKVStore) BatchSet([]infra.KVPair) error { return nil } +func (f *fakeKVStore) Close() error { return nil } + +func (f *fakeKVStore) GetWithOptions(string, *api.QueryOptions) (string, error) { + return f.getVal, f.getErr +} + +func TestGetLatestBlock_MissingKeyReturnsZeroNoError(t *testing.T) { + bs := NewBlockStore(&fakeKVStore{getErr: kvstore.ErrKeyNotFound}) + latest, err := bs.GetLatestBlock("ETH") + require.NoError(t, err) + assert.Equal(t, uint64(0), latest) +} + +func TestGetLatestBlock_StoreErrorPropagates(t *testing.T) { + storeErr := errors.New("redis down") + bs := NewBlockStore(&fakeKVStore{getErr: storeErr}) + _, err := bs.GetLatestBlock("ETH") + require.ErrorIs(t, err, storeErr) +} + +func TestGetLatestBlock_ParsesValue(t *testing.T) { + bs := NewBlockStore(&fakeKVStore{getVal: "12345"}) + latest, err := bs.GetLatestBlock("ETH") + require.NoError(t, err) + assert.Equal(t, uint64(12345), latest) +} From 3101c43197688710f8025bb16325385f6c0c293a Mon Sep 17 00:00:00 2001 From: vietddude Date: Thu, 18 Jun 2026 17:28:25 +0700 Subject: [PATCH 2/2] fix: remove start block for simplicity --- internal/worker/regular.go | 196 ++++++++++++++++---------------- internal/worker/regular_test.go | 47 +++----- 2 files changed, 119 insertions(+), 124 deletions(-) diff --git a/internal/worker/regular.go b/internal/worker/regular.go index e9a281d..ec93d59 100644 --- a/internal/worker/regular.go +++ b/internal/worker/regular.go @@ -118,116 +118,107 @@ func (rw *RegularWorker) processRegularBlocks() error { return nil } - end := min(rw.currentBlock+uint64(rw.config.Throttle.BatchSize)-1, latest) - rw.logger.Info( - "Processing range", - "chain", - rw.chain.GetName(), - "start", - rw.currentBlock, - "end", - end, - "size", - end-rw.currentBlock+1, - ) - - // Store original range for logging - originalStart := rw.currentBlock - originalEnd := end + start := rw.currentBlock + end := min(start+uint64(rw.config.Throttle.BatchSize)-1, latest) startTime := time.Now() + rw.logger.Info("Processing range", + "chain", rw.chain.GetName(), + "start", start, "end", end, "size", end-start+1, + ) - results, err := rw.chain.GetBlocks(rw.ctx, rw.currentBlock, end, rw.config.Throttle.Parallel) + results, err := rw.chain.GetBlocks(rw.ctx, start, end, rw.config.Throttle.Parallel) if err != nil { return fmt.Errorf("get blocks: %w", err) } - lastSuccess := rw.currentBlock - 1 - var lastSuccessHash string - var ( - processErr error - stopTick bool - ) - if rw.isReorgCheckRequired() { - stopTick, processErr = rw.processReorgCheckedBatch(results, end, &lastSuccess, &lastSuccessHash) - } else { - for _, res := range results { - if rw.handleBlockResult(res) { - lastSuccess = res.Number - lastSuccessHash = res.Block.Hash - } - } - } - + lastSuccess, lastSuccessHash, stopTick, processErr := rw.processBatch(results, end) if stopTick { return nil } - indexedAt := time.Time{} - if lastSuccess >= rw.currentBlock { - rw.currentBlock = lastSuccess + 1 - _ = rw.blockStore.SaveLatestBlock(rw.chain.GetNetworkInternalCode(), lastSuccess) - indexedAt = time.Now().UTC() - - if lastSuccessHash != "" { - rw.addBlockHash(lastSuccess, lastSuccessHash) - } - } + indexedAt := rw.commitProgress(lastSuccess, lastSuccessHash) rw.updateHeadStatus(latest, indexedAt) rw.logger.Info("Processed latest blocks", "chain", rw.chain.GetName(), - "start", originalStart, - "end", originalEnd, + "start", start, "end", end, "elapsed", time.Since(startTime), "last_success", lastSuccess, - "expected", originalEnd-originalStart+1, + "expected", end-start+1, "got", len(results), ) return processErr } -// determineStartingBlock picks a start block from the chain head and the last -// indexed block. GetLatestBlock returns (0, nil) for a cold start, so a non-nil -// kvErr means the store is genuinely down: we must never rewind to config.StartBlock. -func (rw *RegularWorker) determineStartingBlock() uint64 { - chainLatest, chainErr := rw.getLatestBlockWithRetry() - kvLatest, kvErr := rw.blockStore.GetLatestBlock(rw.chain.GetNetworkInternalCode()) +// processBatch applies a batch of fetched blocks, using reorg-checked processing +// for chains that need it. It returns the last successfully indexed block and its +// hash, plus stop=true when the tick should end early (e.g. a reorg rollback). +func (rw *RegularWorker) processBatch( + results []indexer.BlockResult, + end uint64, +) (lastSuccess uint64, lastSuccessHash string, stop bool, err error) { + lastSuccess = rw.currentBlock - 1 - switch { - case kvErr != nil && chainErr == nil: - rw.logger.Error("Block store unavailable, starting from chain head to avoid stale rewind", - "chain", rw.chain.GetName(), "chainLatest", chainLatest, "error", kvErr) - return chainLatest + if rw.isReorgCheckRequired() { + stop, err = rw.processReorgCheckedBatch(results, end, &lastSuccess, &lastSuccessHash) + return lastSuccess, lastSuccessHash, stop, err + } - case kvErr != nil: - rw.logger.Error("Chain RPC and block store both unreachable after retries; using config.StartBlock as last resort", - "chain", rw.chain.GetName(), "startBlock", rw.config.StartBlock, "error", kvErr) - return uint64(rw.config.StartBlock) + for _, res := range results { + if rw.handleBlockResult(res) { + lastSuccess = res.Number + lastSuccessHash = res.Block.Hash + } + } + return lastSuccess, lastSuccessHash, false, nil +} - case kvLatest == 0 && chainErr == nil: // cold start, chain reachable - return chainLatest +// commitProgress advances currentBlock past the last indexed block, persisting +// the checkpoint and its hash. Returns the indexing timestamp, or zero if no new +// block was indexed. +func (rw *RegularWorker) commitProgress(lastSuccess uint64, lastSuccessHash string) time.Time { + if lastSuccess < rw.currentBlock { + return time.Time{} + } + rw.currentBlock = lastSuccess + 1 + _ = rw.blockStore.SaveLatestBlock(rw.chain.GetNetworkInternalCode(), lastSuccess) + if lastSuccessHash != "" { + rw.addBlockHash(lastSuccess, lastSuccessHash) + } + return time.Now().UTC() +} - case kvLatest == 0: - rw.logger.Warn("Cold start and chain RPC unavailable, using config.StartBlock", - "chain", rw.chain.GetName(), "startBlock", rw.config.StartBlock) - return uint64(rw.config.StartBlock) +// determineStartingBlock picks a start block from the last indexed block and the +// chain head. GetLatestBlock returns (0, nil) for a cold start, so a non-nil +// kvErr means the store is genuinely down. When there is no prior block to +// resume from, the chain head is the only safe anchor and we wait for it. +func (rw *RegularWorker) determineStartingBlock() uint64 { + kvLatest, kvErr := rw.blockStore.GetLatestBlock(rw.chain.GetNetworkInternalCode()) - case chainErr != nil: - rw.logger.Warn("Chain RPC failed, resuming from KV latest", - "chain", rw.chain.GetName(), "kvLatest", kvLatest) - return kvLatest + // Resume from the last indexed block when the store has one. + if kvErr == nil && kvLatest > 0 { + chainLatest, chainErr := rw.getLatestBlockWithRetry() + if chainErr != nil { + rw.logger.Warn("Chain RPC failed, resuming from KV latest", + "chain", rw.chain.GetName(), "kvLatest", kvLatest) + return kvLatest + } + if chainLatest > kvLatest { + ranges := rw.queueCatchupRanges(kvLatest+1, chainLatest) + rw.logger.Info("Queued catchup ranges", + "chain", rw.chain.GetName(), + "gap", fmt.Sprintf("%d-%d", kvLatest+1, chainLatest), + "ranges_created", len(ranges), + ) + } + return chainLatest } - // Both available: resume from KV, queuing any gap up to chain head. - if chainLatest > kvLatest { - ranges := rw.queueCatchupRanges(kvLatest+1, chainLatest) - rw.logger.Info("Queued catchup ranges", - "chain", rw.chain.GetName(), - "gap", fmt.Sprintf("%d-%d", kvLatest+1, chainLatest), - "ranges_created", len(ranges), - ) + if kvErr != nil { + rw.logger.Error("Block store unavailable, starting from chain head", + "chain", rw.chain.GetName(), "error", kvErr) } - return chainLatest + return rw.waitForChainHead() } // queueCatchupRanges splits [start, end] into chunks, persists them, and @@ -250,22 +241,44 @@ func (rw *RegularWorker) queueCatchupRanges(start, end uint64) []blockstore.Catc return ranges } -// getLatestBlockWithRetry fetches the chain head, retrying transient RPC -// failures so a momentary provider hiccup at startup doesn't force a fallback. +// getLatestBlockWithRetry fetches the chain head with bounded retries, used when +// we already have a KV block to fall back to. func (rw *RegularWorker) getLatestBlockWithRetry() (uint64, error) { - delay := rw.startBlockRetryDelay - if delay <= 0 { - delay = startBlockRetryDelay - } var latest uint64 err := retry.Constant(func() error { var err error latest, err = rw.chain.GetLatestBlockNumber(rw.ctx) return err - }, delay, startBlockRetryAttempts) + }, rw.startBlockDelay(), startBlockRetryAttempts) return latest, err } +// waitForChainHead blocks until the chain head can be fetched, retrying with +// backoff. It is the cold-start anchor when no prior block exists in the store. +// Returns 0 if the context is cancelled while waiting. +func (rw *RegularWorker) waitForChainHead() uint64 { + for { + latest, err := rw.chain.GetLatestBlockNumber(rw.ctx) + if err == nil { + return latest + } + rw.logger.Warn("Waiting for chain head before starting", + "chain", rw.chain.GetName(), "error", err) + select { + case <-rw.ctx.Done(): + return 0 + case <-time.After(rw.startBlockDelay()): + } + } +} + +func (rw *RegularWorker) startBlockDelay() time.Duration { + if rw.startBlockRetryDelay > 0 { + return rw.startBlockRetryDelay + } + return startBlockRetryDelay +} + func (rw *RegularWorker) detectAndHandleReorg(res *indexer.BlockResult) (bool, error) { prevNum := res.Block.Number - 1 storedHash := rw.getBlockHash(prevNum) @@ -597,13 +610,6 @@ func (rw *RegularWorker) fetchRegularBlock(blockNumber uint64) (indexer.BlockRes }, nil } -func checkContinuity(prev, curr indexer.BlockResult) bool { - if prev.Error != nil || curr.Error != nil || prev.Block == nil || curr.Block == nil { - return false - } - return prev.Block.Hash == curr.Block.ParentHash -} - func blockResultError(res indexer.BlockResult) string { if res.Error != nil { return res.Error.Message diff --git a/internal/worker/regular_test.go b/internal/worker/regular_test.go index d91a6b8..dad74de 100644 --- a/internal/worker/regular_test.go +++ b/internal/worker/regular_test.go @@ -118,16 +118,6 @@ func TestRegularWorkerProcessRegularBlocksMarksUnresolvedGapFailed(t *testing.T) require.Equal(t, []uint64{100, 100}, chain.getBlockCalls) } -func TestCheckContinuityReturnsFalseForNilBlocks(t *testing.T) { - t.Parallel() - - require.False(t, checkContinuity(indexer.BlockResult{}, indexer.BlockResult{})) - require.False(t, checkContinuity( - indexer.BlockResult{Block: &types.Block{Hash: "0x100"}}, - indexer.BlockResult{}, - )) -} - func TestBaseWorkerExecuteRecoverableConvertsPanicToError(t *testing.T) { t.Parallel() @@ -176,55 +166,50 @@ func TestRegularWorkerDetermineStartingBlockUpdatesCatchupRegistry(t *testing.T) func TestRegularWorkerDetermineStartingBlockStoreDownChainUp(t *testing.T) { t.Parallel() - // Store is genuinely down but the chain is reachable: must start from chain - // head, never rewind to config.StartBlock. + // Store down but chain reachable: anchor on the chain head. chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 500} store := &stubBlockStore{getLatestBlockErr: errors.New("redis down")} rw := newTestRegularWorker(chain, store, 0, 2) - rw.config.StartBlock = 100 require.Equal(t, uint64(500), rw.determineStartingBlock()) } -func TestRegularWorkerDetermineStartingBlockStoreAndChainDown(t *testing.T) { +func TestRegularWorkerDetermineStartingBlockWaitsForChainHead(t *testing.T) { t.Parallel() - // Both store and chain are down after retries: config.StartBlock is the - // explicit last resort. - chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latestErr: errors.New("tls: internal error")} - store := &stubBlockStore{getLatestBlockErr: errors.New("redis down")} + // Cold start with chain failing transiently: wait until the head responds. + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 500, latestFailN: 2} + store := &stubBlockStore{latestBlock: 0} rw := newTestRegularWorker(chain, store, 0, 2) - rw.config.StartBlock = 100 rw.startBlockRetryDelay = time.Millisecond - require.Equal(t, uint64(100), rw.determineStartingBlock()) + require.Equal(t, uint64(500), rw.determineStartingBlock()) } func TestRegularWorkerDetermineStartingBlockColdStartChainUp(t *testing.T) { t.Parallel() - // Store reachable with no prior block (cold start) and chain reachable: - // start from chain head, not config.StartBlock. + // Cold start with chain reachable: start from chain head. chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 500} store := &stubBlockStore{latestBlock: 0} rw := newTestRegularWorker(chain, store, 0, 2) - rw.config.StartBlock = 100 require.Equal(t, uint64(500), rw.determineStartingBlock()) } -func TestRegularWorkerDetermineStartingBlockColdStartChainDown(t *testing.T) { +func TestRegularWorkerDetermineStartingBlockChainCancelledReturnsZero(t *testing.T) { t.Parallel() - // Genuine cold start (no prior block) and chain unavailable: fall back to - // config.StartBlock. + // Cold start, chain never responds and context is cancelled: return 0. chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latestErr: errors.New("tls: internal error")} store := &stubBlockStore{latestBlock: 0} rw := newTestRegularWorker(chain, store, 0, 2) - rw.config.StartBlock = 100 rw.startBlockRetryDelay = time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + cancel() + rw.ctx = ctx - require.Equal(t, uint64(100), rw.determineStartingBlock()) + require.Equal(t, uint64(0), rw.determineStartingBlock()) } func TestRegularWorkerDetermineStartingBlockChainDownResumesFromKV(t *testing.T) { @@ -234,7 +219,6 @@ func TestRegularWorkerDetermineStartingBlockChainDownResumesFromKV(t *testing.T) chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latestErr: errors.New("tls: internal error")} store := &stubBlockStore{latestBlock: 42} rw := newTestRegularWorker(chain, store, 0, 2) - rw.config.StartBlock = 100 rw.startBlockRetryDelay = time.Millisecond require.Equal(t, uint64(42), rw.determineStartingBlock()) @@ -321,6 +305,7 @@ type stubIndexer struct { networkType enum.NetworkType latest uint64 latestErr error + latestFailN int // fail GetLatestBlockNumber this many times before succeeding getBlocksFunc func(ctx context.Context, from, to uint64, isParallel bool) ([]indexer.BlockResult, error) getBlockFunc func(ctx context.Context, number uint64) (*types.Block, error) getBlockCalls []uint64 @@ -339,6 +324,10 @@ func (s *stubIndexer) GetNetworkInternalCode() string { } func (s *stubIndexer) GetLatestBlockNumber(context.Context) (uint64, error) { + if s.latestFailN > 0 { + s.latestFailN-- + return 0, errors.New("tls: internal error") + } if s.latestErr != nil { return 0, s.latestErr }