diff --git a/config.md b/config.md index 4dd0802..4631aba 100644 --- a/config.md +++ b/config.md @@ -137,8 +137,8 @@ |Key|Description|Type|Default Value| |---|-----------|----|-------------| -|enabled|When true, transaction receipts fetched during canonical chain build are cached in memory for reuse|`boolean`|`false` -|size|Maximum of transaction receipts to hold in the receipt cache|`int`|`5000` +|enabled|When true, transaction receipts fetched during canonical chain build are cached in memory and served for receipt queries.|`boolean`|`false` +|size|Maximum number of transaction receipts to hold in the receipt cache|`int`|`5000` ## connector.retry diff --git a/internal/msgs/en_config_descriptions.go b/internal/msgs/en_config_descriptions.go index 3b4b29e..66224c6 100644 --- a/internal/msgs/en_config_descriptions.go +++ b/internal/msgs/en_config_descriptions.go @@ -32,8 +32,8 @@ var ( _ = ffc("config.connector.dataFormat", "Configure the JSON data format for query output and events", "map,flat_array,self_describing") _ = ffc("config.connector.gasEstimationFactor", "The factor to apply to the gas estimation to determine the gas limit", i18n.FloatType) _ = ffc("config.connector.blockCacheSize", "Maximum of blocks to hold in the block info cache", i18n.IntType) - _ = ffc("config.connector.receiptCache.enabled", "When true, transaction receipts fetched during canonical chain build are cached in memory for reuse", i18n.BooleanType) - _ = ffc("config.connector.receiptCache.size", "Maximum of transaction receipts to hold in the receipt cache", i18n.IntType) + _ = ffc("config.connector.receiptCache.enabled", "When true, transaction receipts fetched during canonical chain build are cached in memory and served for receipt queries.", i18n.BooleanType) + _ = ffc("config.connector.receiptCache.size", "Maximum number of transaction receipts to hold in the receipt cache", i18n.IntType) _ = ffc("config.connector.blockPollingInterval", "Interval for polling to check for new blocks", i18n.TimeDurationType) _ = ffc("config.connector.queryLoopRetry.initialDelay", "Initial delay for retrying query requests to the RPC endpoint, applicable to all the query loops", i18n.TimeDurationType) _ = ffc("config.connector.queryLoopRetry.factor", "Factor to increase the delay by, between each query request retry to the RPC endpoint, applicable to all the query loops", i18n.FloatType) diff --git a/pkg/ethblocklistener/block_receipt_fetcher.go b/pkg/ethblocklistener/block_receipt_fetcher.go index 49bcf86..e905a68 100644 --- a/pkg/ethblocklistener/block_receipt_fetcher.go +++ b/pkg/ethblocklistener/block_receipt_fetcher.go @@ -106,25 +106,71 @@ func (brr *blockReceiptRequest) run() { } // resetReceiptCache clears all cached transaction receipts when the canonical chain -// changes (fork trim, chain rebuild, or re-initialization). Receipts are keyed only by -// transaction hash, so after a reorg the same hash can refer to a block that is no -// longer canonical. Purging avoids serving stale receipts. +// is re-initialized with no valid existing block. Receipts are keyed only by transaction +// hash, so after a reorg the same hash can refer to a block that is no longer canonical. // // The generation counter invalidates any in-flight async fetches that complete after // the reset, so they cannot repopulate the cache with orphaned data. // -// The current implementation is intentionally brute-force. -// All receipts are dropped and refetched for the whole canonical chain at once. -// Selective invalidation by block height or hash could be more efficient and is a -// candidate for future enhancement if re-orgs are frequent enough to justify the complexity. +// Fork trims use invalidateReceiptsForForkTrim instead, which drops only receipts for +// the orphaned tail blocks and leaves the canonical prefix cache intact. func (bl *blockListener) resetReceiptCache() { if bl.txReceiptCache == nil { return } bl.txReceiptCacheLock.Lock() defer bl.txReceiptCacheLock.Unlock() + bl.resetReceiptCacheLocked() +} + +// Caller MUST hold txReceiptCacheLock +func (bl *blockListener) resetReceiptCacheLocked() { bl.txReceiptCache.Purge() bl.txReceiptCacheGeneration++ + bl.invalidatedReceiptBlockHashes = make(map[string]struct{}) +} + +// invalidateReceiptsForForkTrim drops cached receipts for blocks removed from the +// canonical chain tail during a fork. Blocks indexed by the listener carry their +// transaction hashes, so entries are removed directly by tx hash. Trimmed block +// hashes are also recorded so in-flight async fetches cannot repopulate them. +func (bl *blockListener) invalidateReceiptsForForkTrim(trimmedBlocks []*ethrpc.BlockInfoJSONRPC) { + if bl.txReceiptCache == nil || len(trimmedBlocks) == 0 { + return + } + + bl.txReceiptCacheLock.Lock() + defer bl.txReceiptCacheLock.Unlock() + + // Check if the invalidated block hash set is too large, if so reset the cache + // monitored head length, which is the deepest possible single fork trim. Entries are + // only removed when a trimmed block becomes canonical again, so on a long-running + // listener the set would otherwise grow with every fork trim. Hitting the bound + // falls back to a full cache reset, which is always safe. + // When no valid block was found during a rebuild, the cache is fully reset anyway, so + // that path is not affected by this check. + if len(bl.invalidatedReceiptBlockHashes)+len(trimmedBlocks) > bl.MonitoredHeadLength { + bl.resetReceiptCacheLocked() + return + } + for _, bi := range trimmedBlocks { + if bi == nil || bi.Hash == nil { + continue + } + bl.invalidatedReceiptBlockHashes[bi.Hash.String()] = struct{}{} + for _, txHash := range bi.Transactions { + bl.txReceiptCache.Remove(txHash.String()) + } + } +} + +func (bl *blockListener) revalidateReceiptBlockHash(blockHash ethtypes.HexBytes0xPrefix) { + if bl.txReceiptCache == nil || blockHash == nil { + return + } + bl.txReceiptCacheLock.Lock() + delete(bl.invalidatedReceiptBlockHashes, blockHash.String()) + bl.txReceiptCacheLock.Unlock() } func (bl *blockListener) getReceiptCacheGeneration() uint64 { @@ -133,7 +179,11 @@ func (bl *blockListener) getReceiptCacheGeneration() uint64 { return bl.txReceiptCacheGeneration } -func (bl *blockListener) storeReceiptsInCache(receipts []*ethrpc.TxReceiptJSONRPC, generation uint64) { +// storeReceiptsInCache stores the receipts in the cache. +// The generation must match the current generation, otherwise the receipts are ignored. +// The blockHash is used to check if the receipts are for a block that is no longer canonical. +// If the blockHash is provided, the receipts are only stored if the block is not in the invalidatedReceiptBlockHashes map. +func (bl *blockListener) storeReceiptsInCache(receipts []*ethrpc.TxReceiptJSONRPC, generation uint64, blockHash ethtypes.HexBytes0xPrefix) { if bl.txReceiptCache == nil { return } @@ -142,8 +192,18 @@ func (bl *blockListener) storeReceiptsInCache(receipts []*ethrpc.TxReceiptJSONRP if generation != bl.txReceiptCacheGeneration { return } + if blockHash != nil { + if _, invalidated := bl.invalidatedReceiptBlockHashes[blockHash.String()]; invalidated { + return + } + } for _, r := range receipts { if r != nil && r.TransactionHash != nil { + if r.BlockHash != nil { + if _, invalidated := bl.invalidatedReceiptBlockHashes[r.BlockHash.String()]; invalidated { + continue + } + } bl.txReceiptCache.Add(r.TransactionHash.String(), r) } } @@ -162,27 +222,43 @@ func (bl *blockListener) getCachedTransactionReceipt(txHash string) (*ethrpc.TxR return cached.(*ethrpc.TxReceiptJSONRPC), true } -func (bl *blockListener) fetchAndCacheBlockReceipts(bi *ethrpc.BlockInfoJSONRPC) { +type pendingReceiptFetch struct { + blockInfo *ethrpc.BlockInfoJSONRPC + generation uint64 +} + +// queueReceiptFetch records a block whose receipts should be fetched into the cache once +// the canonical chain lock is released. The block is canonical at this point, so its hash +// is revalidated and the cache generation snapshotted here, under the lock. If a fork trim +// removes the block again before the fetch completes, the hash is re-invalidated and +// storeReceiptsInCache discards the results. +// +// Caller MUST hold the canonicalChain WRITE LOCK +func (bl *blockListener) queueReceiptFetch(bi *ethrpc.BlockInfoJSONRPC) { if bl.txReceiptCache == nil { return } - generation := bl.getReceiptCacheGeneration() - bl.FetchBlockReceiptsAsync(bi.Number.Uint64(), bi.Hash, func(receipts []*ethrpc.TxReceiptJSONRPC, err error) { - if err != nil { - log.L(bl.ctx).Debugf("Failed to fetch receipts for block %d / %s: %v", bi.Number.Uint64(), bi.Hash, err) - return - } - bl.storeReceiptsInCache(receipts, generation) + bl.revalidateReceiptBlockHash(bi.Hash) + bl.pendingReceiptFetches = append(bl.pendingReceiptFetches, &pendingReceiptFetch{ + blockInfo: bi, + generation: bl.getReceiptCacheGeneration(), }) } -func (bl *blockListener) refetchReceiptsForCanonicalChain() { - if bl.txReceiptCache == nil { - return - } - for pos := bl.canonicalChain.Front(); pos != nil; pos = pos.Next() { - if pos.Value != nil { - bl.fetchAndCacheBlockReceipts(pos.Value.(*ethrpc.BlockInfoJSONRPC)) - } +// dispatchPendingReceiptFetches starts the async receipt fetches for blocks queued by +// queueReceiptFetch. FetchBlockReceiptsAsync blocks when the fetch concurrency throttle +// is saturated, so this must be called WITHOUT the canonical chain lock held, to avoid +// stalling readers of the chain. +func (bl *blockListener) dispatchPendingReceiptFetches(pending []*pendingReceiptFetch) { + for _, f := range pending { + bi := f.blockInfo + generation := f.generation + bl.FetchBlockReceiptsAsync(bi.Number.Uint64(), bi.Hash, func(receipts []*ethrpc.TxReceiptJSONRPC, err error) { + if err != nil { + log.L(bl.ctx).Debugf("Failed to fetch receipts for block %d / %s: %v", bi.Number.Uint64(), bi.Hash, err) + return + } + bl.storeReceiptsInCache(receipts, generation, bi.Hash) + }) } } diff --git a/pkg/ethblocklistener/block_receipt_fetcher_test.go b/pkg/ethblocklistener/block_receipt_fetcher_test.go index fd78c04..46cc56a 100644 --- a/pkg/ethblocklistener/block_receipt_fetcher_test.go +++ b/pkg/ethblocklistener/block_receipt_fetcher_test.go @@ -19,6 +19,7 @@ package ethblocklistener import ( "context" "testing" + "time" "github.com/hyperledger-firefly/common/pkg/fftypes" "github.com/hyperledger-firefly/common/pkg/i18n" @@ -28,6 +29,7 @@ import ( "github.com/hyperledger-firefly/signer/pkg/rpcbackend" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" ) func TestFetchBlockReceiptsAsyncOptimizedOk(t *testing.T) { @@ -214,7 +216,7 @@ func TestGetTransactionReceiptUsesCache(t *testing.T) { BlockHash: generateTestHash(1977), } - bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{cachedReceipt}, bl.getReceiptCacheGeneration()) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{cachedReceipt}, bl.getReceiptCacheGeneration(), cachedReceipt.BlockHash) receipt, err := bl.GetTransactionReceipt(context.Background(), txHash) assert.NoError(t, err) @@ -229,7 +231,7 @@ func TestResetReceiptCacheClearsCachedReceipts(t *testing.T) { txHash := "0x6197ef1a58a2a592bb447efb651f0db7945de21aa8048801b250bd7b7431f9b6" bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash)}, - }, bl.getReceiptCacheGeneration()) + }, bl.getReceiptCacheGeneration(), nil) _, ok := bl.getCachedTransactionReceipt(txHash) assert.True(t, ok) @@ -250,7 +252,7 @@ func TestStoreReceiptsInCacheIgnoresStaleGeneration(t *testing.T) { txHash := "0x6197ef1a58a2a592bb447efb651f0db7945de21aa8048801b250bd7b7431f9b6" bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash)}, - }, gen) + }, gen, nil) _, ok := bl.getCachedTransactionReceipt(txHash) assert.False(t, ok) @@ -268,7 +270,7 @@ func TestReconcileConfirmationsForTransactionUsesCachedReceipt(t *testing.T) { BlockNumber: ethtypes.HexUint64(1977), BlockHash: generateTestHash(1977), }, - }, bl.getReceiptCacheGeneration()) + }, bl.getReceiptCacheGeneration(), generateTestHash(1977)) mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByNumber", "0x7b9", false).Return(nil).Run(func(args mock.Arguments) { *args[1].(**ethrpc.EVMBlockWithTxHashesJSONRPC) = ðrpc.EVMBlockWithTxHashesJSONRPC{BlockHeaderJSONRPC: ethrpc.BlockHeaderJSONRPC{ @@ -300,13 +302,13 @@ func TestReceiptCacheEvictsWhenFull(t *testing.T) { bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash1)}, - }, gen) + }, gen, nil) bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash2)}, - }, gen) + }, gen, nil) bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash3)}, - }, gen) + }, gen, nil) _, ok := bl.getCachedTransactionReceipt(txHash1) assert.False(t, ok) @@ -315,3 +317,217 @@ func TestReceiptCacheEvictsWhenFull(t *testing.T) { _, ok = bl.getCachedTransactionReceipt(txHash3) assert.True(t, ok) } + +func TestInvalidateReceiptsForForkTrimRemovesOnlyTrimmedBlocks(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + gen := bl.getReceiptCacheGeneration() + txHash100 := "0x1000000000000000000000000000000000000000000000000000000000000001" + txHash101 := "0x1010000000000000000000000000000000000000000000000000000000000001" + txHash102 := "0x1020000000000000000000000000000000000000000000000000000000000001" + blockHash100 := generateTestHash(100) + blockHash101 := generateTestHash(101) + blockHash102 := generateTestHash(102) + + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash100), + BlockHash: blockHash100, + }}, gen, blockHash100) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash101), + BlockHash: blockHash101, + }}, gen, blockHash101) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash102), + BlockHash: blockHash102, + }}, gen, blockHash102) + + bl.invalidateReceiptsForForkTrim([]*ethrpc.BlockInfoJSONRPC{ + {Hash: blockHash101, Number: 101, Transactions: []ethtypes.HexBytes0xPrefix{ethtypes.MustNewHexBytes0xPrefix(txHash101)}}, + {Hash: blockHash102, Number: 102, Transactions: []ethtypes.HexBytes0xPrefix{ethtypes.MustNewHexBytes0xPrefix(txHash102)}}, + }) + + _, ok := bl.getCachedTransactionReceipt(txHash100) + assert.True(t, ok) + _, ok = bl.getCachedTransactionReceipt(txHash101) + assert.False(t, ok) + _, ok = bl.getCachedTransactionReceipt(txHash102) + assert.False(t, ok) + assert.Equal(t, gen, bl.getReceiptCacheGeneration()) +} + +func TestInvalidateReceiptsForForkTrimFallsBackToResetAtBound(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + gen := bl.getReceiptCacheGeneration() + txHash := "0x1000000000000000000000000000000000000000000000000000000000000001" + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash), + }}, gen, nil) + + trimmed := make([]*ethrpc.BlockInfoJSONRPC, bl.MonitoredHeadLength+1) + for i := range trimmed { + trimmed[i] = ðrpc.BlockInfoJSONRPC{Hash: generateTestHash(uint64(i)), Number: ethtypes.HexUint64(i)} + } + bl.invalidateReceiptsForForkTrim(trimmed) + + // The whole cache is reset rather than tracking an unbounded invalidation set + assert.Equal(t, gen+1, bl.getReceiptCacheGeneration()) + assert.Empty(t, bl.invalidatedReceiptBlockHashes) + _, ok := bl.getCachedTransactionReceipt(txHash) + assert.False(t, ok) +} + +func TestStoreReceiptsInCacheIgnoresInvalidatedForkTrimBlock(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + blockHash := generateTestHash(102) + txHash := "0x1020000000000000000000000000000000000000000000000000000000000001" + bl.invalidateReceiptsForForkTrim([]*ethrpc.BlockInfoJSONRPC{ + {Hash: blockHash, Number: 102}, + }) + + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash), + BlockHash: blockHash, + }}, bl.getReceiptCacheGeneration(), blockHash) + + _, ok := bl.getCachedTransactionReceipt(txHash) + assert.False(t, ok) +} + +func TestRevalidateReceiptBlockHashAllowsCachingAgain(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + blockHash := generateTestHash(101) + txHash := "0x1010000000000000000000000000000000000000000000000000000000000001" + bl.invalidateReceiptsForForkTrim([]*ethrpc.BlockInfoJSONRPC{ + { + Hash: blockHash, + Number: 101, + Transactions: []ethtypes.HexBytes0xPrefix{ethtypes.MustNewHexBytes0xPrefix(txHash)}, + }, + }) + + bl.revalidateReceiptBlockHash(blockHash) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash), + BlockHash: blockHash, + }}, bl.getReceiptCacheGeneration(), blockHash) + + _, ok := bl.getCachedTransactionReceipt(txHash) + assert.True(t, ok) +} + +func TestQueuedReceiptFetchRevalidatesPreviouslyInvalidatedBlock(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t, func(conf *BlockListenerConfig, mRPC *rpcbackendmocks.Backend, cancelCtx context.CancelFunc) { + conf.UseGetBlockReceipts = true + }) + defer done() + + blockHash := generateTestHash(101) + txHash := "0x1010000000000000000000000000000000000000000000000000000000000001" + bl.invalidateReceiptsForForkTrim([]*ethrpc.BlockInfoJSONRPC{ + { + Hash: blockHash, + Number: 101, + Transactions: []ethtypes.HexBytes0xPrefix{ethtypes.MustNewHexBytes0xPrefix(txHash)}, + }, + }) + + receipt := ðrpc.TxReceiptJSONRPC{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash), + BlockHash: blockHash, + } + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockReceipts", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + *args[1].(*[]*ethrpc.TxReceiptJSONRPC) = []*ethrpc.TxReceiptJSONRPC{receipt} + }) + + bl.canonicalChainLock.Lock() + bl.queueReceiptFetch(ðrpc.BlockInfoJSONRPC{ + Number: 101, + Hash: blockHash, + }) + pending := bl.pendingReceiptFetches + bl.pendingReceiptFetches = nil + bl.canonicalChainLock.Unlock() + bl.dispatchPendingReceiptFetches(pending) + + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if _, ok := bl.getCachedTransactionReceipt(txHash); ok { + break + } + time.Sleep(5 * time.Millisecond) + } + + _, ok := bl.getCachedTransactionReceipt(txHash) + assert.True(t, ok) +} + +func TestHandleNewBlockForkTrimPreservesPrefixReceiptCache(t *testing.T) { + _, bl, _, done := newTestBlockListener(t, func(conf *BlockListenerConfig, mRPC *rpcbackendmocks.Backend, cancelCtx context.CancelFunc) { + conf.UseGetBlockReceipts = true + }) + defer done() + + bl.canonicalChain = createTestChain(100, 102) + gen := bl.getReceiptCacheGeneration() + txHash100 := "0x1000000000000000000000000000000000000000000000000000000000000001" + txHash101 := "0x1010000000000000000000000000000000000000000000000000000000000001" + txHash102 := "0x1020000000000000000000000000000000000000000000000000000000000001" + blockHash100 := generateTestHash(100) + blockHash101 := generateTestHash(101) + blockHash102 := generateTestHash(102) + + for pos := bl.canonicalChain.Front(); pos != nil; pos = pos.Next() { + bi := pos.Value.(*ethrpc.BlockInfoJSONRPC) + switch bi.Number.Uint64() { + case 100: + bi.Transactions = []ethtypes.HexBytes0xPrefix{ethtypes.MustNewHexBytes0xPrefix(txHash100)} + case 101: + bi.Transactions = []ethtypes.HexBytes0xPrefix{ethtypes.MustNewHexBytes0xPrefix(txHash101)} + case 102: + bi.Transactions = []ethtypes.HexBytes0xPrefix{ethtypes.MustNewHexBytes0xPrefix(txHash102)} + } + } + + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash100), + BlockHash: blockHash100, + }}, gen, blockHash100) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash101), + BlockHash: blockHash101, + }}, gen, blockHash101) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash102), + BlockHash: blockHash102, + }}, gen, blockHash102) + + forkBlock := ðrpc.BlockInfoJSONRPC{ + Number: 101, + Hash: generateTestHash(101, 1), + ParentHash: blockHash100, + } + + bl.canonicalChainLock.Lock() + bl.handleNewBlock(forkBlock, bl.canonicalChain.Front()) + // The receipts of the new fork block are queued for fetching, not dispatched under the lock + require.Len(t, bl.pendingReceiptFetches, 1) + assert.Equal(t, forkBlock, bl.pendingReceiptFetches[0].blockInfo) + bl.pendingReceiptFetches = nil + bl.canonicalChainLock.Unlock() + + _, ok := bl.getCachedTransactionReceipt(txHash100) + assert.True(t, ok) + _, ok = bl.getCachedTransactionReceipt(txHash101) + assert.False(t, ok) + _, ok = bl.getCachedTransactionReceipt(txHash102) + assert.False(t, ok) + assert.Equal(t, gen, bl.getReceiptCacheGeneration()) +} diff --git a/pkg/ethblocklistener/blocklistener.go b/pkg/ethblocklistener/blocklistener.go index 93964ea..6cbc453 100644 --- a/pkg/ethblocklistener/blocklistener.go +++ b/pkg/ethblocklistener/blocklistener.go @@ -137,8 +137,10 @@ type blockListener struct { headBlockInfo *ethrpc.BlockInfoJSONRPC // full info for the current head block, when seen // tx receipts indexed during canonical chain build, keyed by transaction hash - txReceiptCacheLock sync.RWMutex - txReceiptCacheGeneration uint64 + txReceiptCacheLock sync.RWMutex + txReceiptCacheGeneration uint64 + invalidatedReceiptBlockHashes map[string]struct{} + pendingReceiptFetches []*pendingReceiptFetch // queued under the canonicalChain write lock, dispatched after it is released // headBlockNumber mode: last head value sent on the block listener channel (only written from listenLoop) currentChainHead uint64 @@ -190,6 +192,7 @@ func NewBlockListenerSupplyBackend(ctx context.Context, retry *retry.Retry, conf if err != nil { return nil, i18n.WrapError(ctx, err, msgs.MsgCacheInitFail, "receipt") } + bl.invalidatedReceiptBlockHashes = make(map[string]struct{}) } return bl, nil } @@ -487,8 +490,19 @@ func (bl *blockListener) listenLoop() { // work backwards building a new view and notify about all blocks that are changed in that process. func (bl *blockListener) reconcileCanonicalChain(bi *ethrpc.BlockInfoJSONRPC) *list.Element { bl.canonicalChainLock.Lock() - defer bl.canonicalChainLock.Unlock() + pos := bl.reconcileCanonicalChainLocked(bi) + pending := bl.pendingReceiptFetches + bl.pendingReceiptFetches = nil + bl.canonicalChainLock.Unlock() + + // Dispatch queued receipt fetches only after releasing the lock - dispatch blocks when + // the fetch concurrency throttle is saturated, and must not stall readers of the chain + bl.dispatchPendingReceiptFetches(pending) + return pos +} +// Caller MUST hold the canonicalChain WRITE LOCK +func (bl *blockListener) reconcileCanonicalChainLocked(bi *ethrpc.BlockInfoJSONRPC) *list.Element { bl.checkAndSetHighestBlock(bi) // Find the position of this block in the block sequence @@ -534,7 +548,6 @@ func (bl *blockListener) handleNewBlock(mbi *ethrpc.BlockInfoJSONRPC, addAfter * // Ok, we can add this block var newElem *list.Element - forkTrim := false if addAfter == nil { bl.resetReceiptCache() _ = bl.canonicalChain.Init() @@ -544,12 +557,18 @@ func (bl *blockListener) handleNewBlock(mbi *ethrpc.BlockInfoJSONRPC, addAfter * // Trim everything from this point onwards. Note that the following cases are covered on other paths: // - This was just a duplicate notification of a block that fits into our chain - discarded in reconcileCanonicalChain() // - There was a gap before us in the chain, and the tail is still valid - we would have called rebuildCanonicalChain() above + var trimmedBlocks []*ethrpc.BlockInfoJSONRPC nextElem := newElem.Next() for nextElem != nil { toRemove := nextElem nextElem = nextElem.Next() + if toRemove.Value != nil { + trimmedBlocks = append(trimmedBlocks, toRemove.Value.(*ethrpc.BlockInfoJSONRPC)) + } _ = bl.canonicalChain.Remove(toRemove) - forkTrim = true + } + if len(trimmedBlocks) > 0 { + bl.invalidateReceiptsForForkTrim(trimmedBlocks) } } @@ -557,11 +576,7 @@ func (bl *blockListener) handleNewBlock(mbi *ethrpc.BlockInfoJSONRPC, addAfter * for bl.canonicalChain.Len() > bl.MonitoredHeadLength { _ = bl.canonicalChain.Remove(bl.canonicalChain.Front()) } - - if forkTrim { - bl.resetReceiptCache() - } - bl.fetchAndCacheBlockReceipts(mbi) + bl.queueReceiptFetch(mbi) log.L(bl.ctx).Debugf("Added block %d / %s parent=%s to in-memory canonical chain (new length=%d)", mbi.Number.Uint64(), mbi.Hash, mbi.ParentHash, bl.canonicalChain.Len()) @@ -574,10 +589,8 @@ func (bl *blockListener) handleNewBlock(mbi *ethrpc.BlockInfoJSONRPC, addAfter * // // Caller MUST hold the canonicalChain WRITE LOCK func (bl *blockListener) rebuildCanonicalChain() *list.Element { - bl.resetReceiptCache() // If none of our blocks were valid, start from the first block number we've notified about previously lastValidBlock := bl.trimToLastValidBlock() - bl.refetchReceiptsForCanonicalChain() var nextBlockNumber uint64 var expectedParentHash ethtypes.HexBytes0xPrefix if lastValidBlock != nil { @@ -585,6 +598,8 @@ func (bl *blockListener) rebuildCanonicalChain() *list.Element { log.L(bl.ctx).Infof("Canonical chain partially rebuilding from block %d", nextBlockNumber) expectedParentHash = lastValidBlock.Hash } else { + // no valid block found, so we need to re-initialize the chain + bl.resetReceiptCache() firstBlock := bl.canonicalChain.Front() if firstBlock == nil || firstBlock.Value == nil { return nil @@ -629,7 +644,7 @@ func (bl *blockListener) rebuildCanonicalChain() *list.Element { } bl.checkAndSetHighestBlock(bi) - bl.fetchAndCacheBlockReceipts(bi) + bl.queueReceiptFetch(bi) } return notifyPos @@ -640,6 +655,7 @@ func (bl *blockListener) trimToLastValidBlock() (lastValidBlock *ethrpc.BlockInf // First remove from the end until we get a block that matches the current un-cached query view from the chain lastElem := bl.canonicalChain.Back() var startingNumber *uint64 + var trimmedBlocks []*ethrpc.BlockInfoJSONRPC for lastElem != nil && lastElem.Value != nil { // Query the block that is no at this blockNumber @@ -671,8 +687,12 @@ func (bl *blockListener) trimToLastValidBlock() (lastValidBlock *ethrpc.BlockInf } break } + trimmedBlocks = append(trimmedBlocks, currentViewBlock) lastElem = lastElem.Prev() } + // Invalidate cached receipts for the trimmed blocks in one pass. If no valid block was + // found the caller resets the whole cache anyway, but invalidating here is still safe. + bl.invalidateReceiptsForForkTrim(trimmedBlocks) if startingNumber != nil && lastValidBlock != nil && *startingNumber != lastValidBlock.Number.Uint64() { log.L(bl.ctx).Debugf("Canonical chain trimmed from block %d to block %d (total number of in memory blocks: %d)", startingNumber, lastValidBlock.Number.Uint64(), bl.MonitoredHeadLength) diff --git a/pkg/ethblocklistener/blocklistener_test.go b/pkg/ethblocklistener/blocklistener_test.go index affffda..cd8517c 100644 --- a/pkg/ethblocklistener/blocklistener_test.go +++ b/pkg/ethblocklistener/blocklistener_test.go @@ -244,24 +244,23 @@ func TestBlockListenerConstructorFailCacheConfig(t *testing.T) { } func TestBlockListenerStartGettingHighestBlockRetry(t *testing.T) { + testLatch := newTestLatch() _, bl, mRPC, done := newTestBlockListener(t) mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber"). Return(&rpcbackend.RPCError{Message: "pop"}).Once() mockInitialBlockHeight(mRPC, 12345) - mockSeedBlockNotFound(mRPC, 12345-(50-1)).Maybe() - mockNewBlockFilter(mRPC, testBlockFilterID1).Maybe() - mockFilterChangesEmpty(mRPC).Maybe() + mockSeedBlockNotFound(mRPC, 12345-(50-1)).Once() + mockNewBlockFilter(mRPC, testBlockFilterID1).Once() + mockFilterChangesEmpty(mRPC, testLatch.complete).Once() h, ok := bl.GetHighestBlock(bl.ctx) assert.Equal(t, uint64(12345), h) assert.True(t, ok) - done() - - <-bl.listenLoopDone - mRPC.AssertExpectations(t) + testLatch.waitComplete() + done() } func TestBlockListenerStartGettingHighestBlockFailBeforeStop(t *testing.T) { @@ -755,6 +754,66 @@ func TestTrimToLastValidBlockRemovesInvalidTail(t *testing.T) { mRPC.AssertExpectations(t) } +// TestTrimToLastValidBlockInvalidatesTrimmedReceiptsOnly covers the receipt cache wiring of a +// partial rebuild. Receipts for the surviving prefix stay cached, receipts for the trimmed +// stale tail are invalidated, and the cache generation is unchanged (no full reset). +func TestTrimToLastValidBlockInvalidatesTrimmedReceiptsOnly(t *testing.T) { + h99 := testBlockHashFor(99) + h100 := testBlockHashFor(100) + h101Stale := testBlockHashFor(101, 999) + h102Stale := testBlockHashFor(102, 999) + h101 := testBlockHashFor(101) + h102 := testBlockHashFor(102) + + _, bl, mRPC, done := newTestBlockListener(t) + defer done() + + mockBlockByNumber(mRPC, 102, &h102).Once() + mockBlockByNumber(mRPC, 101, &h101).Once() + mockBlockByNumber(mRPC, 100, &h100).Once() + + tx100 := ethtypes.MustNewHexBytes0xPrefix("0x1000000000000000000000000000000000000000000000000000000000000001") + tx101 := ethtypes.MustNewHexBytes0xPrefix("0x1010000000000000000000000000000000000000000000000000000000000001") + tx102 := ethtypes.MustNewHexBytes0xPrefix("0x1020000000000000000000000000000000000000000000000000000000000001") + + gen := bl.getReceiptCacheGeneration() + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ + {TransactionHash: tx100, BlockHash: h100}, + {TransactionHash: tx101, BlockHash: h101Stale}, + {TransactionHash: tx102, BlockHash: h102Stale}, + }, gen, nil) + + b100 := ðrpc.BlockInfoJSONRPC{Number: ethtypes.HexUint64(100), Hash: h100, ParentHash: h99, Transactions: []ethtypes.HexBytes0xPrefix{tx100}} + b101 := ðrpc.BlockInfoJSONRPC{Number: ethtypes.HexUint64(101), Hash: h101Stale, ParentHash: h100, Transactions: []ethtypes.HexBytes0xPrefix{tx101}} + b102 := ðrpc.BlockInfoJSONRPC{Number: ethtypes.HexUint64(102), Hash: h102Stale, ParentHash: h101Stale, Transactions: []ethtypes.HexBytes0xPrefix{tx102}} + + bl.canonicalChainLock.Lock() + bl.canonicalChain.PushBack(b100) + bl.canonicalChain.PushBack(b101) + bl.canonicalChain.PushBack(b102) + lastValid := bl.trimToLastValidBlock() + bl.canonicalChainLock.Unlock() + + require.NotNil(t, lastValid) + require.Equal(t, uint64(100), lastValid.Number.Uint64()) + + // Prefix receipt survives, trimmed tail receipts are gone, and no full cache reset happened + _, ok := bl.getCachedTransactionReceipt(tx100.String()) + assert.True(t, ok) + _, ok = bl.getCachedTransactionReceipt(tx101.String()) + assert.False(t, ok) + _, ok = bl.getCachedTransactionReceipt(tx102.String()) + assert.False(t, ok) + assert.Equal(t, gen, bl.getReceiptCacheGeneration()) + + // A late async fetch result for a trimmed block must be discarded + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{{TransactionHash: tx101, BlockHash: h101Stale}}, gen, h101Stale) + _, ok = bl.getCachedTransactionReceipt(tx101.String()) + assert.False(t, ok) + + mRPC.AssertExpectations(t) +} + func TestBlockListenerReorgReplaceTail(t *testing.T) { block1001Hash := testBlockHashFor(1001)