diff --git a/configs/config.example.yaml b/configs/config.example.yaml index 17bb4de..dc219d7 100644 --- a/configs/config.example.yaml +++ b/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" diff --git a/internal/indexer/aptos_test.go b/internal/indexer/aptos_test.go index 32bba6b..a7c83ae 100644 --- a/internal/indexer/aptos_test.go +++ b/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 diff --git a/internal/indexer/cosmos_test.go b/internal/indexer/cosmos_test.go index 2a81464..c6d3261 100644 --- a/internal/indexer/cosmos_test.go +++ b/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) diff --git a/internal/indexer/evm.go b/internal/indexer/evm.go index 2e9af4e..82920eb 100644 --- a/internal/indexer/evm.go +++ b/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, diff --git a/internal/indexer/evm_test.go b/internal/indexer/evm_test.go index f7b86a4..a121091 100644 --- a/internal/indexer/evm_test.go +++ b/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" diff --git a/internal/indexer/stellar_test.go b/internal/indexer/stellar_test.go index 52abd5c..0b43ee1 100644 --- a/internal/indexer/stellar_test.go +++ b/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 diff --git a/internal/indexer/xrp_test.go b/internal/indexer/xrp_test.go index 6bd07c6..78ed7cb 100644 --- a/internal/indexer/xrp_test.go +++ b/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 } diff --git a/internal/worker/base.go b/internal/worker/base.go index a36fb8a..cc4a655 100644 --- a/internal/worker/base.go +++ b/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() diff --git a/internal/worker/base_test.go b/internal/worker/base_test.go index ce52aa3..33caee1 100644 --- a/internal/worker/base_test.go +++ b/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 } diff --git a/internal/worker/catchup.go b/internal/worker/catchup.go index fe3f6ac..f544d4b 100644 --- a/internal/worker/catchup.go +++ b/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, ) diff --git a/internal/worker/catchup_test.go b/internal/worker/catchup_test.go index a5ddf81..183f330 100644 --- a/internal/worker/catchup_test.go +++ b/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) +} diff --git a/internal/worker/regular.go b/internal/worker/regular.go index ec93d59..72243db 100644 --- a/internal/worker/regular.go +++ b/internal/worker/regular.go @@ -101,25 +101,35 @@ func (rw *RegularWorker) processRegularBlocks() error { if err != nil { return fmt.Errorf("get latest block: %w", err) } + processingHead := rw.processingHead(latest) rw.updateHeadStatus(latest, time.Time{}) - rw.logger.Info("Got latest block", "latest", latest, "current", rw.currentBlock) + rw.logger.Info("Got latest block", + "latest", latest, + "processing_head", processingHead, + "confirmations", rw.config.Confirmations, + "current", rw.currentBlock, + ) // Lag detection: if we're too far behind, jump to chain head and queue skipped range for catchup - if rw.skipAheadIfLagging(latest) { + if rw.skipAheadIfLagging(processingHead) { rw.updateHeadStatus(latest, time.Time{}) return nil } - if rw.currentBlock > latest { - rw.logger.Info("Waiting for new blocks...", "current", rw.currentBlock, "latest", latest) + if rw.currentBlock > processingHead { + rw.logger.Info("Waiting for confirmed blocks...", + "current", rw.currentBlock, + "latest", latest, + "processing_head", processingHead, + ) time.Sleep(rw.config.PollInterval) rw.updateHeadStatus(latest, time.Time{}) return nil } start := rw.currentBlock - end := min(start+uint64(rw.config.Throttle.BatchSize)-1, latest) + end := min(start+uint64(rw.config.Throttle.BatchSize)-1, processingHead) startTime := time.Now() rw.logger.Info("Processing range", "chain", rw.chain.GetName(), @@ -203,15 +213,19 @@ func (rw *RegularWorker) determineStartingBlock() uint64 { "chain", rw.chain.GetName(), "kvLatest", kvLatest) return kvLatest } - if chainLatest > kvLatest { - ranges := rw.queueCatchupRanges(kvLatest+1, chainLatest) + processingHead := rw.processingHead(chainLatest) + if processingHead > kvLatest { + ranges := rw.queueCatchupRanges(kvLatest+1, processingHead) rw.logger.Info("Queued catchup ranges", "chain", rw.chain.GetName(), - "gap", fmt.Sprintf("%d-%d", kvLatest+1, chainLatest), + "gap", fmt.Sprintf("%d-%d", kvLatest+1, processingHead), "ranges_created", len(ranges), ) + return processingHead } - return chainLatest + // A newly enabled confirmation delay can put the processing head behind + // the persisted checkpoint. Never rewind and re-emit those blocks. + return kvLatest } if kvErr != nil { @@ -260,7 +274,7 @@ func (rw *RegularWorker) waitForChainHead() uint64 { for { latest, err := rw.chain.GetLatestBlockNumber(rw.ctx) if err == nil { - return latest + return rw.processingHead(latest) } rw.logger.Warn("Waiting for chain head before starting", "chain", rw.chain.GetName(), "error", err) diff --git a/internal/worker/regular_test.go b/internal/worker/regular_test.go index 8852084..859b821 100644 --- a/internal/worker/regular_test.go +++ b/internal/worker/regular_test.go @@ -80,6 +80,55 @@ func TestRegularWorkerProcessRegularBlocksRecoversGapViaGetBlock(t *testing.T) { require.Equal(t, []uint64{101, 102}, chain.getBlockCalls) } +func TestRegularWorkerProcessRegularBlocksStopsAtConfirmedHead(t *testing.T) { + t.Parallel() + + var gotFrom, gotTo uint64 + chain := &stubIndexer{ + name: "bsc", + internalCode: "BSC_MAINNET", + networkType: enum.NetworkTypeEVM, + latest: 102, + getBlocksFunc: func(_ context.Context, from, to uint64, _ bool) ([]indexer.BlockResult, error) { + gotFrom, gotTo = from, to + return []indexer.BlockResult{{ + Number: 100, + Block: &types.Block{Number: 100, Hash: "0x100", ParentHash: "0x099"}, + }}, nil + }, + } + store := &stubBlockStore{} + rw := newTestRegularWorker(chain, store, 100, 20) + rw.config.Confirmations = 2 + + err := rw.processRegularBlocks() + require.NoError(t, err) + require.Equal(t, uint64(100), gotFrom) + require.Equal(t, uint64(100), gotTo) + require.Equal(t, uint64(101), rw.currentBlock) + require.Equal(t, []uint64{100}, store.savedLatest) +} + +func TestRegularWorkerProcessRegularBlocksWaitsWhenConfirmedHeadIsBehind(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{ + name: "bsc", + internalCode: "BSC_MAINNET", + networkType: enum.NetworkTypeEVM, + latest: 100, + getBlocksFunc: func(context.Context, uint64, uint64, bool) ([]indexer.BlockResult, error) { + t.Fatal("GetBlocks must not be called ahead of the confirmed head") + return nil, nil + }, + } + rw := newTestRegularWorker(chain, &stubBlockStore{}, 91, 20) + rw.config.Confirmations = 10 + + require.NoError(t, rw.processRegularBlocks()) + require.Equal(t, uint64(91), rw.currentBlock) +} + func TestRegularWorkerProcessRegularBlocksMarksUnresolvedGapFailed(t *testing.T) { t.Parallel() @@ -198,6 +247,41 @@ func TestRegularWorkerDetermineStartingBlockColdStartChainUp(t *testing.T) { require.Equal(t, uint64(500), rw.determineStartingBlock()) } +func TestRegularWorkerDetermineStartingBlockUsesConfirmedHead(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{name: "bsc", internalCode: "BSC_MAINNET", networkType: enum.NetworkTypeEVM, latest: 100} + store := &stubBlockStore{latestBlock: 80} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.Confirmations = 10 + + require.Equal(t, uint64(90), rw.determineStartingBlock()) + require.Equal(t, []blockstore.CatchupRange{{Start: 81, End: 90, Current: 80}}, store.savedCatchupRanges) +} + +func TestRegularWorkerDetermineStartingBlockDoesNotRewindPastCheckpoint(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{name: "bsc", internalCode: "BSC_MAINNET", networkType: enum.NetworkTypeEVM, latest: 100} + store := &stubBlockStore{latestBlock: 95} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.Confirmations = 10 + + require.Equal(t, uint64(95), rw.determineStartingBlock()) + require.Empty(t, store.savedCatchupRanges) +} + +func TestRegularWorkerDetermineStartingBlockConfirmedHeadDoesNotUnderflow(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{name: "bsc", internalCode: "BSC_MAINNET", networkType: enum.NetworkTypeEVM, latest: 5} + store := &stubBlockStore{} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.Confirmations = 10 + + require.Equal(t, uint64(0), rw.determineStartingBlock()) +} + func TestRegularWorkerDetermineStartingBlockChainCancelledReturnsZero(t *testing.T) { t.Parallel() diff --git a/pkg/addressbloomfilter/bloomfilter.go b/pkg/addressbloomfilter/bloomfilter.go index 84ca69c..4136dd9 100644 --- a/pkg/addressbloomfilter/bloomfilter.go +++ b/pkg/addressbloomfilter/bloomfilter.go @@ -25,6 +25,9 @@ type WalletAddressBloomFilter interface { // Contains checks if a given address exists in the bloom filter for the specified type. Contains(address string, addressType enum.NetworkType) bool + // ContainsBatch checks multiple addresses in a single call, returning a result per address in order. + ContainsBatch(addresses []string, addressType enum.NetworkType) []bool + // Clear deletes the bloom filter for a given address type. Clear(addressType enum.NetworkType) diff --git a/pkg/addressbloomfilter/inmemory.go b/pkg/addressbloomfilter/inmemory.go index 6046616..5244d81 100644 --- a/pkg/addressbloomfilter/inmemory.go +++ b/pkg/addressbloomfilter/inmemory.go @@ -119,6 +119,18 @@ func (abf *addressBloomFilter) Contains(address string, addressType enum.Network return bf.filter.Test([]byte(address)) } +func (abf *addressBloomFilter) ContainsBatch(addresses []string, addressType enum.NetworkType) []bool { + bf := abf.getOrCreateFilter(addressType) + bf.mu.RLock() + defer bf.mu.RUnlock() + + results := make([]bool, len(addresses)) + for i, address := range addresses { + results[i] = bf.filter.Test([]byte(address)) + } + return results +} + func (abf *addressBloomFilter) Clear(addressType enum.NetworkType) { bf := abf.getOrCreateFilter(addressType) bf.mu.Lock() diff --git a/pkg/addressbloomfilter/redis.go b/pkg/addressbloomfilter/redis.go index 9293b76..f5a1351 100644 --- a/pkg/addressbloomfilter/redis.go +++ b/pkg/addressbloomfilter/redis.go @@ -184,6 +184,43 @@ func (rbf *redisBloomFilter) Contains(address string, addressType enum.NetworkTy return result } +func (rbf *redisBloomFilter) ContainsBatch(addresses []string, addressType enum.NetworkType) []bool { + results := make([]bool, len(addresses)) + if len(addresses) == 0 { + return results + } + + rbf.mu.RLock() + defer rbf.mu.RUnlock() + + key := rbf.getKey(addressType) + client := rbf.redisClient.GetClient() + + args := make([]any, 0, len(addresses)+2) + args = append(args, "BF.MEXISTS", key) + for _, addr := range addresses { + args = append(args, addr) + } + + res, err := client.Do(rbf.ctx, args...).Result() + if err != nil { + logger.Error("Error checking Redis bloom filter batch", "error", err) + return results + } + + items, ok := res.([]any) + if !ok || len(items) != len(addresses) { + logger.Error("Unexpected BF.MEXISTS result shape", "result", res) + return results + } + for i, item := range items { + if n, ok := item.(int64); ok { + results[i] = n == 1 + } + } + return results +} + func (rbf *redisBloomFilter) Clear(addressType enum.NetworkType) { rbf.mu.Lock() defer rbf.mu.Unlock() diff --git a/pkg/store/pubkeystore/store.go b/pkg/store/pubkeystore/store.go index 227ea63..3b6a92b 100644 --- a/pkg/store/pubkeystore/store.go +++ b/pkg/store/pubkeystore/store.go @@ -13,6 +13,7 @@ func composeKey(addressType enum.NetworkType, publicKey string) string { type Store interface { Exist(addressType enum.NetworkType, publicKey string) bool + ExistBatch(addressType enum.NetworkType, publicKeys []string) []bool Save(addressType enum.NetworkType, publicKey string) error Close() error } @@ -34,6 +35,13 @@ func (s *publicKeyStore) Exist(addressType enum.NetworkType, publicKey string) b return s.bloomFilter.Contains(publicKey, addressType) } +func (s *publicKeyStore) ExistBatch(addressType enum.NetworkType, publicKeys []string) []bool { + if s.bloomFilter == nil { + return make([]bool, len(publicKeys)) + } + return s.bloomFilter.ContainsBatch(publicKeys, addressType) +} + func (s *publicKeyStore) Save(addressType enum.NetworkType, publicKey string) error { if s.bloomFilter != nil { s.bloomFilter.Add(publicKey, addressType)