fix: delay indexing by configured confirmations

configs/config.example.yaml

@@ -90,6 +90,7 @@ chains:
     type: "evm"
     start_block: 59040717
     reorg_rollback_window: 30 # custom rollback depth for BNB
+    confirmations: 10 # process 10 blocks behind the RPC head so receipts/logs are stable
     nodes:
       - url: "https://bsc-rpc.publicnode.com"
       - url: "https://bsc.blockrazor.xyz"

internal/indexer/aptos_test.go

@@ -29,6 +29,14 @@ func (m mockAptosPubkeyStore) Exist(_ enum.NetworkType, address string) bool {
 	return ok
 }
 
+func (m mockAptosPubkeyStore) ExistBatch(_ enum.NetworkType, addresses []string) []bool {
+	results := make([]bool, len(addresses))
+	for i, addr := range addresses {
+		_, results[i] = m.addresses[addr]
+	}
+	return results
+}
+
 type aptosRealTransferFixture struct {
 	name          string
 	version       uint64

internal/indexer/cosmos_test.go

@@ -25,6 +25,14 @@ func (m mockCosmosPubkeyStore) Exist(_ enum.NetworkType, address string) bool {
 	return ok
 }
 
+func (m mockCosmosPubkeyStore) ExistBatch(_ enum.NetworkType, addresses []string) []bool {
+	results := make([]bool, len(addresses))
+	for i, addr := range addresses {
+		_, results[i] = m.addresses[addr]
+	}
+	return results
+}
+
 func TestCosmosConvertBlock_ParsesTransfersAndFee(t *testing.T) {
 	txPayload := []byte("tx-one")
 	txEncoded := base64.StdEncoding.EncodeToString(txPayload)

internal/indexer/evm.go

@@ -48,6 +48,7 @@ func (e *EVMIndexer) traceModeActive() bool {
 // PubkeyStore interface for checking if an address is monitored
 type PubkeyStore interface {
 	Exist(addressType enum.NetworkType, address string) bool
+	ExistBatch(addressType enum.NetworkType, addresses []string) []bool
 }
 
 func NewEVMIndexer(chainName string, config config.ChainConfig, failover *rpc.Failover[evm.EthereumAPI], traceFailover *rpc.Failover[evm.EthereumAPI], pubkeyStore PubkeyStore) *EVMIndexer {
@@ -461,6 +462,17 @@ func (e *EVMIndexer) queryERC20TransfersToMonitoredAddresses(ctx context.Context
 	// When two_way_indexing is enabled, both 'to' and 'from' sides are considered.
 	matchedTxHashes := make(map[string]bool)
 
+	type transferAddresses struct {
+		txHash      string
+		fromAddress string
+		toAddress   string
+	}
+	transfers := make([]transferAddresses, 0, len(logs))
+	// Collected once per range and checked via a single ContainsBatch/BF.MEXISTS
+	// call instead of one bloom-filter round trip per log (costly on the Redis backend).
+	seenAddresses := make(map[string]bool, len(logs)*2)
+	addresses := make([]string, 0, len(logs)*2)
+
 	for _, log := range logs {
 		if len(log.Topics) < 3 {
 			logger.Warn("Transfer log has less than 3 topics",
@@ -484,17 +496,38 @@ func (e *EVMIndexer) queryERC20TransfersToMonitoredAddresses(ctx context.Context
 		}
 
 		// Get last 40 hex chars (20 bytes) and add 0x prefix.
-		fromAddress := "0x" + fromAddressTopic[len(fromAddressTopic)-40:]
-		toAddress := "0x" + toAddressTopic[len(toAddressTopic)-40:]
+		fromAddress := evm.ToChecksumAddress("0x" + fromAddressTopic[len(fromAddressTopic)-40:])
+		toAddress := evm.ToChecksumAddress("0x" + toAddressTopic[len(toAddressTopic)-40:])
+
+		transfers = append(transfers, transferAddresses{
+			txHash:      log.TransactionHash,
+			fromAddress: fromAddress,
+			toAddress:   toAddress,
+		})
+		if e.config.TwoWayIndexing && !seenAddresses[fromAddress] {
+			seenAddresses[fromAddress] = true
+			addresses = append(addresses, fromAddress)
+		}
+		if !seenAddresses[toAddress] {
+			seenAddresses[toAddress] = true
+			addresses = append(addresses, toAddress)
+		}
+	}
 
-		fromMonitored := e.config.TwoWayIndexing && e.pubkeyStore.Exist(enum.NetworkTypeEVM, evm.ToChecksumAddress(fromAddress))
-		toMonitored := e.pubkeyStore.Exist(enum.NetworkTypeEVM, evm.ToChecksumAddress(toAddress))
+	existsByAddress := make(map[string]bool, len(addresses))
+	for i, exists := range e.pubkeyStore.ExistBatch(enum.NetworkTypeEVM, addresses) {
+		existsByAddress[addresses[i]] = exists
+	}
+
+	for _, t := range transfers {
+		fromMonitored := e.config.TwoWayIndexing && existsByAddress[t.fromAddress]
+		toMonitored := existsByAddress[t.toAddress]
 		if fromMonitored || toMonitored {
-			matchedTxHashes[log.TransactionHash] = true
+			matchedTxHashes[t.txHash] = true
 			logger.Info("MATCHED ERC20 TRANSFER",
-				"tx_hash", log.TransactionHash,
-				"from_address", fromAddress,
-				"to_address", toAddress,
+				"tx_hash", t.txHash,
+				"from_address", t.fromAddress,
+				"to_address", t.toAddress,
 			)
 		}
 	}
@@ -526,11 +559,34 @@ func (e *EVMIndexer) extractReceiptTxHashes(blocks map[uint64]*evm.Block, traceA
 		}
 	}
 
+	// candidateTx holds a tx whose match decision needs an address lookup,
+	// deferred so all addresses across every block in this batch can be
+	// resolved with a single ContainsBatch/ExistBatch call.
+	type candidateTx struct {
+		blockNum    uint64
+		hash        string
+		isSafe      bool
+		toAddress   string
+		fromAddress string // checked only when TwoWayIndexing is enabled
+	}
+	var candidates []candidateTx
+	blockSeen := make(map[uint64]map[string]bool) // shared dedup set per block, reused across both passes
+	seenAddresses := make(map[string]bool)
+	var addresses []string
+	collectAddress := func(addr string) {
+		if addr == "" || seenAddresses[addr] {
+			return
+		}
+		seenAddresses[addr] = true
+		addresses = append(addresses, addr)
+	}
+
 	for blockNum, block := range blocks {
 		if block == nil {
 			continue
 		}
 		seen := make(map[string]bool) // dedup within this block
+		blockSeen[blockNum] = seen
 		for _, tx := range block.Transactions {
 			totalTxs++
 			isSafeExecution := evm.IsSafeExecTransaction(tx.Input)
@@ -558,12 +614,17 @@ func (e *EVMIndexer) extractReceiptTxHashes(blocks map[uint64]*evm.Block, traceA
 				isNativeTransfer := tx.To != "" && (tx.Input == "" || tx.Input == "0x")
 
 				if isNativeTransfer {
-					toMonitored := e.pubkeyStore.Exist(enum.NetworkTypeEVM, evm.ToChecksumAddress(tx.To))
-					fromMonitored := e.config.TwoWayIndexing && tx.From != "" && e.pubkeyStore.Exist(enum.NetworkTypeEVM, evm.ToChecksumAddress(tx.From))
-					if toMonitored || fromMonitored {
-						nativeTransfers++
-						appendHash(blockNum, tx.Hash, seen)
+					toAddress := evm.ToChecksumAddress(tx.To)
+					var fromAddress string
+					if e.config.TwoWayIndexing && tx.From != "" {
+						fromAddress = evm.ToChecksumAddress(tx.From)
 					}
+					collectAddress(toAddress)
+					collectAddress(fromAddress)
+					candidates = append(candidates, candidateTx{
+						blockNum: blockNum, hash: tx.Hash, isSafe: false,
+						toAddress: toAddress, fromAddress: fromAddress,
+					})
 				}
 			} else {
 				// Filter must match ExtractSafeTransfers acceptance criteria.
@@ -576,12 +637,17 @@ func (e *EVMIndexer) extractReceiptTxHashes(blocks map[uint64]*evm.Block, traceA
 					continue
 				}
 
-				toMonitored := e.pubkeyStore.Exist(enum.NetworkTypeEVM, evm.ToChecksumAddress(params.To))
-				fromMonitored := e.config.TwoWayIndexing && tx.To != "" && e.pubkeyStore.Exist(enum.NetworkTypeEVM, evm.ToChecksumAddress(tx.To))
-				if toMonitored || fromMonitored {
-					safeTransfers++
-					appendHash(blockNum, tx.Hash, seen)
+				toAddress := evm.ToChecksumAddress(params.To)
+				var fromAddress string
+				if e.config.TwoWayIndexing && tx.To != "" {
+					fromAddress = evm.ToChecksumAddress(tx.To)
 				}
+				collectAddress(toAddress)
+				collectAddress(fromAddress)
+				candidates = append(candidates, candidateTx{
+					blockNum: blockNum, hash: tx.Hash, isSafe: true,
+					toAddress: toAddress, fromAddress: fromAddress,
+				})
 			}
 
 			// When trace mode is active, fetch receipts for ALL contract calls.
@@ -597,6 +663,28 @@ func (e *EVMIndexer) extractReceiptTxHashes(blocks map[uint64]*evm.Block, traceA
 		}
 	}
 
+	if e.pubkeyStore != nil && len(candidates) > 0 {
+		existsByAddress := make(map[string]bool, len(addresses))
+		for i, exists := range e.pubkeyStore.ExistBatch(enum.NetworkTypeEVM, addresses) {
+			existsByAddress[addresses[i]] = exists
+		}
+
+		for _, c := range candidates {
+			seen := blockSeen[c.blockNum]
+
+			toMonitored := existsByAddress[c.toAddress]
+			fromMonitored := c.fromAddress != "" && existsByAddress[c.fromAddress]
+			if toMonitored || fromMonitored {
+				if c.isSafe {
+					safeTransfers++
+				} else {
+					nativeTransfers++
+				}
+				appendHash(c.blockNum, c.hash, seen)
+			}
+		}
+	}
+
 	if e.pubkeyStore != nil {
 		logger.Debug("[SELECTIVE RECEIPTS - NATIVE]",
 			"total_txs", totalTxs,

internal/indexer/evm_test.go

@@ -23,6 +23,14 @@ func (s evmPubkeyStoreStub) Exist(_ enum.NetworkType, address string) bool {
 	return s.addresses[address]
 }
 
+func (s evmPubkeyStoreStub) ExistBatch(_ enum.NetworkType, addresses []string) []bool {
+	results := make([]bool, len(addresses))
+	for i, addr := range addresses {
+		results[i] = s.addresses[addr]
+	}
+	return results
+}
+
 const safeExecInputTest = "0x6a761202000000000000000000000000c26dc13d057824342d5480b153f288bd1c5e3e9d000000000000000000000000000000000000000000000000016345785d8a000000000000000000000000000000000000000000000000000000000000000001400000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001600000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000008200000000000000000000000040a1ade6e21129c09cfb3c501efba5e56ba9a3e50000000000000000000000000000000000000000000000000000000000000000010000000000000000000000005c45ee16de1570b662e32e7960e1e501dd93cde4000000000000000000000000000000000000000000000000000000000000000001a5aa81e56a92b6f46bf78edb6a34e0086c0bb5c1e9d2c20ae60d49c1a8de6b4f15c14847f1ba3f6c35b51dbefae4c74a01bf7c28a0a0c53fb4e73a6d40e36f76b1c00000000000000000000000000000000000000000000000000000000000000"
 
 const ethMainnetRPC = "https://ethereum-rpc.publicnode.com"

internal/indexer/stellar_test.go

@@ -29,6 +29,14 @@ func (m mockStellarPubkeyStore) Exist(_ enum.NetworkType, address string) bool {
 	return ok
 }
 
+func (m mockStellarPubkeyStore) ExistBatch(_ enum.NetworkType, addresses []string) []bool {
+	results := make([]bool, len(addresses))
+	for i, addr := range addresses {
+		_, results[i] = m.addresses[addr]
+	}
+	return results
+}
+
 type mockStellarAPI struct {
 	mu           sync.Mutex
 	latest       uint64

internal/indexer/xrp_test.go

@@ -25,6 +25,14 @@ func (m mockXRPPubkeyStore) Exist(_ enum.NetworkType, address string) bool {
 	return ok
 }
 
+func (m mockXRPPubkeyStore) ExistBatch(_ enum.NetworkType, addresses []string) []bool {
+	results := make([]bool, len(addresses))
+	for i, addr := range addresses {
+		_, results[i] = m.addresses[addr]
+	}
+	return results
+}
+
 type mockXRPAPI struct {
 	ledgers map[uint64]*xrp.Ledger
 }

internal/worker/base.go

@@ -51,6 +51,16 @@ type BaseWorker struct {
 	statusRegistry status.StatusRegistry
 }
 
+// processingHead returns the highest block this worker may process after the
+// chain-specific confirmation delay. The guard prevents uint64 underflow while
+// a chain is still below its configured confirmation count.
+func (bw *BaseWorker) processingHead(latest uint64) uint64 {
+	if latest <= bw.config.Confirmations {
+		return 0
+	}
+	return latest - bw.config.Confirmations
+}
+
 // Stop stops the worker and cleans up internal resources
 func (bw *BaseWorker) Stop() {
 	bw.cancel()

internal/worker/base_test.go

@@ -30,6 +30,14 @@ func (m capturePubkeyStore) Exist(_ enum.NetworkType, address string) bool {
 	return ok
 }
 
+func (m capturePubkeyStore) ExistBatch(_ enum.NetworkType, addresses []string) []bool {
+	results := make([]bool, len(addresses))
+	for i, addr := range addresses {
+		_, results[i] = m.addresses[addr]
+	}
+	return results
+}
+
 func (m capturePubkeyStore) Save(enum.NetworkType, string) error { return nil }
 func (m capturePubkeyStore) Close() error                        { return nil }
 

internal/worker/catchup.go

@@ -174,7 +174,8 @@ func (cw *CatchupWorker) loadCatchupProgress() []blockstore.CatchupRange {
 	// Only create a new range if no existing ranges found
 	if len(ranges) == 0 {
 		if latest, err1 := cw.blockStore.GetLatestBlock(cw.chain.GetNetworkInternalCode()); err1 == nil {
-			if head, err2 := cw.chain.GetLatestBlockNumber(cw.ctx); err2 == nil && head > latest {
+			if chainHead, err2 := cw.chain.GetLatestBlockNumber(cw.ctx); err2 == nil {
+				head := cw.processingHead(chainHead)
 				if head <= latest {
 					// no gap between head and latest
 					return ranges
@@ -183,7 +184,8 @@ func (cw *CatchupWorker) loadCatchupProgress() []blockstore.CatchupRange {
 				cw.logger.Info("Creating new catchup range",
 					"chain", cw.chain.GetName(),
 					"latest_block", latest,
-					"head_block", head,
+					"chain_head", chainHead,
+					"processing_head", head,
 					"catchup_start", start, "catchup_end", end,
 					"blocks_to_catchup", end-latest,
 				)

internal/worker/catchup_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/fystack/multichain-indexer/internal/indexer"
 	"github.com/fystack/multichain-indexer/pkg/common/enum"
 	"github.com/fystack/multichain-indexer/pkg/common/types"
+	"github.com/fystack/multichain-indexer/pkg/store/blockstore"
 	"github.com/stretchr/testify/require"
 )
 
@@ -101,3 +102,21 @@ func TestCatchupWorkerPicksUpRangesQueuedAfterDrain(t *testing.T) {
 		t.Fatal("runCatchup did not stop after context cancel")
 	}
 }
+
+func TestCatchupWorkerLoadProgressStopsAtConfirmedHead(t *testing.T) {
+	t.Parallel()
+
+	chain := &stubIndexer{
+		name:         "bsc",
+		internalCode: "BSC_MAINNET",
+		networkType:  enum.NetworkTypeEVM,
+		latest:       100,
+	}
+	store := &stubBlockStore{latestBlock: 80}
+	cw := newTestCatchupWorker(context.Background(), chain, store)
+	cw.config.Confirmations = 10
+
+	ranges := cw.loadCatchupProgress()
+	require.Equal(t, []blockstore.CatchupRange{{Start: 81, End: 90, Current: 80}}, ranges)
+	require.Equal(t, ranges, store.savedCatchupRanges)
+}