From 6ab6b15a44128d083dcf7ec1ded5dba6d664ce90 Mon Sep 17 00:00:00 2001 From: Chengxuan Xing Date: Mon, 6 Jul 2026 11:06:25 +0100 Subject: [PATCH 1/2] Adding an option to cache txn receipts when building canonical chain Signed-off-by: Chengxuan Xing --- config.md | 7 ++ internal/ethereum/config.go | 4 + internal/ethereum/ethereum.go | 2 + internal/ethereum/ethereum_test.go | 6 + internal/msgs/en_config_descriptions.go | 2 + pkg/ethblocklistener/block_receipt_fetcher.go | 70 +++++++++++ .../block_receipt_fetcher_test.go | 113 ++++++++++++++++++ pkg/ethblocklistener/blocklistener.go | 24 ++++ .../blocklistener_blockquery.go | 3 + pkg/ethblocklistener/blocklistener_test.go | 10 ++ 10 files changed, 241 insertions(+) diff --git a/config.md b/config.md index 7b5ee62..4dd0802 100644 --- a/config.md +++ b/config.md @@ -133,6 +133,13 @@ |initialDelay|Initial delay for retrying query requests to the RPC endpoint, applicable to all the query loops|[`time.Duration`](https://pkg.go.dev/time#Duration)|`100ms` |maxDelay|Maximum delay for between each query request retry to the RPC endpoint, applicable to all the query loops|[`time.Duration`](https://pkg.go.dev/time#Duration)|`30s` +## connector.receiptCache + +|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` + ## connector.retry |Key|Description|Type|Default Value| diff --git a/internal/ethereum/config.go b/internal/ethereum/config.go index d13cd9c..ceb0eee 100644 --- a/internal/ethereum/config.go +++ b/internal/ethereum/config.go @@ -27,6 +27,8 @@ const ( ConfigDataFormat = "dataFormat" BlockPollingInterval = "blockPollingInterval" BlockCacheSize = "blockCacheSize" + ReceiptCacheEnabled = "receiptCache.enabled" + ReceiptCacheSize = "receiptCache.size" ChainTrackingMode = "chainTrackingMode" EventsCatchupPageSize = "events.catchupPageSize" EventsCatchupThreshold = "events.catchupThreshold" @@ -71,6 +73,8 @@ func InitConfig(conf config.Section) { wsclient.InitConfig(conf) conf.AddKnownKey(WebSocketsEnabled, false) conf.AddKnownKey(BlockCacheSize, 250) + conf.AddKnownKey(ReceiptCacheEnabled, false) + conf.AddKnownKey(ReceiptCacheSize, 5000) conf.AddKnownKey(BlockPollingInterval, "1s") conf.AddKnownKey(ChainTrackingMode, ffcapi.ChainTrackingModeFull) conf.AddKnownKey(ConfigDataFormat, "map") diff --git a/internal/ethereum/ethereum.go b/internal/ethereum/ethereum.go index 62a9095..73ecb26 100644 --- a/internal/ethereum/ethereum.go +++ b/internal/ethereum/ethereum.go @@ -174,6 +174,8 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connecto MonitoredHeadLength: int(c.checkpointBlockGap), HederaCompatibilityMode: conf.GetBool(HederaCompatibilityMode), BlockCacheSize: conf.GetInt(BlockCacheSize), + ReceiptCacheEnabled: conf.GetBool(ReceiptCacheEnabled), + ReceiptCacheSize: conf.GetInt(ReceiptCacheSize), MaxAsyncBlockFetchConcurrency: conf.GetInt(MaxAsyncBlockFetchConcurrency), UseGetBlockReceipts: conf.GetBool(UseGetBlockReceipts), ChainTrackingMode: c.chainTrackingMode, diff --git a/internal/ethereum/ethereum_test.go b/internal/ethereum/ethereum_test.go index 01314ab..7d5230a 100644 --- a/internal/ethereum/ethereum_test.go +++ b/internal/ethereum/ethereum_test.go @@ -147,6 +147,12 @@ func TestConnectorInit(t *testing.T) { assert.Regexp(t, "FF23040", err) conf.Set(TxCacheSize, "1") + conf.Set(ReceiptCacheEnabled, true) + conf.Set(ReceiptCacheSize, "-1") + cc, err = NewEthereumConnector(context.Background(), conf) + assert.Regexp(t, "FF23040", err) + + conf.Set(ReceiptCacheSize, "1") conf.Set(EventsCatchupDownscaleRegex, "[") cc, err = NewEthereumConnector(context.Background(), conf) assert.Regexp(t, "FF23051", err) diff --git a/internal/msgs/en_config_descriptions.go b/internal/msgs/en_config_descriptions.go index aa45f8b..3b4b29e 100644 --- a/internal/msgs/en_config_descriptions.go +++ b/internal/msgs/en_config_descriptions.go @@ -32,6 +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.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 4ff4696..c51f6c9 100644 --- a/pkg/ethblocklistener/block_receipt_fetcher.go +++ b/pkg/ethblocklistener/block_receipt_fetcher.go @@ -104,3 +104,73 @@ func (brr *blockReceiptRequest) run() { // No early return in this function - return must happen by reaching here earlyExit = false } + +func (bl *blockListener) resetReceiptCache() { + if bl.txReceiptCache == nil { + return + } + bl.txReceiptCacheLock.Lock() + defer bl.txReceiptCacheLock.Unlock() + bl.txReceiptCache.Purge() + bl.txReceiptCacheGeneration++ +} + +func (bl *blockListener) getReceiptCacheGeneration() uint64 { + bl.txReceiptCacheLock.RLock() + defer bl.txReceiptCacheLock.RUnlock() + return bl.txReceiptCacheGeneration +} + +func (bl *blockListener) storeReceiptsInCache(receipts []*ethrpc.TxReceiptJSONRPC, generation uint64) { + if bl.txReceiptCache == nil { + return + } + bl.txReceiptCacheLock.Lock() + defer bl.txReceiptCacheLock.Unlock() + if generation != bl.txReceiptCacheGeneration { + return + } + for _, r := range receipts { + if r != nil && r.TransactionHash != nil { + bl.txReceiptCache.Add(r.TransactionHash.String(), r) + } + } +} + +func (bl *blockListener) getCachedTransactionReceipt(txHash string) (*ethrpc.TxReceiptJSONRPC, bool) { + if bl.txReceiptCache == nil { + return nil, false + } + bl.txReceiptCacheLock.RLock() + defer bl.txReceiptCacheLock.RUnlock() + cached, ok := bl.txReceiptCache.Get(txHash) + if !ok { + return nil, false + } + return cached.(*ethrpc.TxReceiptJSONRPC), true +} + +func (bl *blockListener) fetchAndCacheBlockReceipts(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) + }) +} + +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)) + } + } +} diff --git a/pkg/ethblocklistener/block_receipt_fetcher_test.go b/pkg/ethblocklistener/block_receipt_fetcher_test.go index c35e58c..fd78c04 100644 --- a/pkg/ethblocklistener/block_receipt_fetcher_test.go +++ b/pkg/ethblocklistener/block_receipt_fetcher_test.go @@ -202,3 +202,116 @@ func TestFetchBlockReceiptsAsyncNonOptimizedNotFound(t *testing.T) { }) <-fetched } + +func TestGetTransactionReceiptUsesCache(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t) + defer done() + + txHash := "0x6197ef1a58a2a592bb447efb651f0db7945de21aa8048801b250bd7b7431f9b6" + cachedReceipt := ðrpc.TxReceiptJSONRPC{ + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash), + BlockNumber: ethtypes.HexUint64(1977), + BlockHash: generateTestHash(1977), + } + + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{cachedReceipt}, bl.getReceiptCacheGeneration()) + + receipt, err := bl.GetTransactionReceipt(context.Background(), txHash) + assert.NoError(t, err) + assert.Equal(t, cachedReceipt, receipt) + mRPC.AssertNotCalled(t, "CallRPC", mock.Anything, mock.Anything, "eth_getTransactionReceipt", mock.Anything) +} + +func TestResetReceiptCacheClearsCachedReceipts(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + txHash := "0x6197ef1a58a2a592bb447efb651f0db7945de21aa8048801b250bd7b7431f9b6" + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ + {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash)}, + }, bl.getReceiptCacheGeneration()) + + _, ok := bl.getCachedTransactionReceipt(txHash) + assert.True(t, ok) + + bl.resetReceiptCache() + + _, ok = bl.getCachedTransactionReceipt(txHash) + assert.False(t, ok) +} + +func TestStoreReceiptsInCacheIgnoresStaleGeneration(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + gen := bl.getReceiptCacheGeneration() + bl.resetReceiptCache() + + txHash := "0x6197ef1a58a2a592bb447efb651f0db7945de21aa8048801b250bd7b7431f9b6" + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ + {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash)}, + }, gen) + + _, ok := bl.getCachedTransactionReceipt(txHash) + assert.False(t, ok) +} + +func TestReconcileConfirmationsForTransactionUsesCachedReceipt(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t) + defer done() + bl.canonicalChain = createTestChain(1976, 1978) + + txHash := "0x6197ef1a58a2a592bb447efb651f0db7945de21aa8048801b250bd7b7431f9b6" + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ + { + TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash), + BlockNumber: ethtypes.HexUint64(1977), + BlockHash: generateTestHash(1977), + }, + }, bl.getReceiptCacheGeneration()) + + 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{ + Number: 1977, + Hash: generateTestHash(1977), + ParentHash: generateTestHash(1976), + }} + }) + + result, receipt, err := bl.ReconcileConfirmationsForTransaction(context.Background(), txHash, []*ethrpc.MinimalBlockInfo{}, 5) + assert.NoError(t, err) + assert.NotNil(t, result) + assert.NotNil(t, receipt) + assert.Equal(t, ethtypes.HexUint64(1977), receipt.BlockNumber) + mRPC.AssertNotCalled(t, "CallRPC", mock.Anything, mock.Anything, "eth_getTransactionReceipt", mock.Anything) +} + +func TestReceiptCacheEvictsWhenFull(t *testing.T) { + _, bl, _, done := newTestBlockListener(t, func(conf *BlockListenerConfig, mRPC *rpcbackendmocks.Backend, cancelCtx context.CancelFunc) { + conf.ReceiptCacheEnabled = true + conf.ReceiptCacheSize = 2 + }) + defer done() + + gen := bl.getReceiptCacheGeneration() + txHash1 := "0x1111111111111111111111111111111111111111111111111111111111111111" + txHash2 := "0x2222222222222222222222222222222222222222222222222222222222222222" + txHash3 := "0x3333333333333333333333333333333333333333333333333333333333333333" + + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ + {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash1)}, + }, gen) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ + {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash2)}, + }, gen) + bl.storeReceiptsInCache([]*ethrpc.TxReceiptJSONRPC{ + {TransactionHash: ethtypes.MustNewHexBytes0xPrefix(txHash3)}, + }, gen) + + _, ok := bl.getCachedTransactionReceipt(txHash1) + assert.False(t, ok) + _, ok = bl.getCachedTransactionReceipt(txHash2) + assert.True(t, ok) + _, ok = bl.getCachedTransactionReceipt(txHash3) + assert.True(t, ok) +} diff --git a/pkg/ethblocklistener/blocklistener.go b/pkg/ethblocklistener/blocklistener.go index 5862b9d..68213ad 100644 --- a/pkg/ethblocklistener/blocklistener.go +++ b/pkg/ethblocklistener/blocklistener.go @@ -62,6 +62,8 @@ type BlockListenerConfig struct { BlockPollingInterval time.Duration `json:"blockPollingInterval"` HederaCompatibilityMode bool `json:"hederaCompatibilityMode"` BlockCacheSize int `json:"blockCacheSize"` + ReceiptCacheEnabled bool `json:"receiptCacheEnabled"` + ReceiptCacheSize int `json:"receiptCacheSize"` IncludeLogsBloom bool `json:"includeLogsBloom"` UseGetBlockReceipts bool `json:"useGetBlockReceipts"` MaxAsyncBlockFetchConcurrency int `json:"maxAsyncBlockFetchConcurrency"` @@ -121,6 +123,7 @@ type blockListener struct { consumerMux sync.Mutex // covers consumers and listenLoopDone consumers map[fftypes.UUID]*BlockUpdateConsumer blockCache *lru.Cache + txReceiptCache *lru.Cache blockFetchConcurrencyThrottle chan *blockReceiptRequest BlockListenerConfig @@ -132,6 +135,10 @@ type blockListener struct { highestBlock uint64 highestBlockGasLimit *ethtypes.HexInteger + // tx receipts indexed during canonical chain build, keyed by transaction hash + txReceiptCacheLock sync.RWMutex + txReceiptCacheGeneration uint64 + // headBlockNumber mode: last head value sent on the block listener channel (only written from listenLoop) currentChainHead uint64 } @@ -177,6 +184,12 @@ func NewBlockListenerSupplyBackend(ctx context.Context, retry *retry.Retry, conf if err != nil { return nil, i18n.WrapError(ctx, err, msgs.MsgCacheInitFail, "block") } + if conf.ReceiptCacheEnabled { + bl.txReceiptCache, err = lru.New(conf.ReceiptCacheSize) + if err != nil { + return nil, i18n.WrapError(ctx, err, msgs.MsgCacheInitFail, "receipt") + } + } return bl, nil } @@ -520,7 +533,9 @@ 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() newElem = bl.canonicalChain.PushBack(mbi) } else { @@ -533,6 +548,7 @@ func (bl *blockListener) handleNewBlock(mbi *ethrpc.BlockInfoJSONRPC, addAfter * toRemove := nextElem nextElem = nextElem.Next() _ = bl.canonicalChain.Remove(toRemove) + forkTrim = true } } @@ -541,6 +557,11 @@ func (bl *blockListener) handleNewBlock(mbi *ethrpc.BlockInfoJSONRPC, addAfter * _ = bl.canonicalChain.Remove(bl.canonicalChain.Front()) } + if forkTrim { + bl.resetReceiptCache() + } + bl.fetchAndCacheBlockReceipts(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()) return newElem @@ -552,8 +573,10 @@ 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 { @@ -605,6 +628,7 @@ func (bl *blockListener) rebuildCanonicalChain() *list.Element { } bl.checkAndSetHighestBlock(bi.Number.Uint64(), bi.GasLimit) + bl.fetchAndCacheBlockReceipts(bi) } return notifyPos diff --git a/pkg/ethblocklistener/blocklistener_blockquery.go b/pkg/ethblocklistener/blocklistener_blockquery.go index 69f4c19..5b241f3 100644 --- a/pkg/ethblocklistener/blocklistener_blockquery.go +++ b/pkg/ethblocklistener/blocklistener_blockquery.go @@ -57,6 +57,9 @@ func (bl *blockListener) getReceiptAndBlock(ctx context.Context, txHash string) } func (bl *blockListener) GetTransactionReceipt(ctx context.Context, txHash string) (ethReceipt *ethrpc.TxReceiptJSONRPC, err error) { + if receipt, ok := bl.getCachedTransactionReceipt(txHash); ok { + return receipt, nil + } rpcErr := bl.backend.CallRPC(ctx, ðReceipt, "eth_getTransactionReceipt", txHash) if rpcErr != nil || ethReceipt == nil { var err error diff --git a/pkg/ethblocklistener/blocklistener_test.go b/pkg/ethblocklistener/blocklistener_test.go index 256e4f8..3b6916d 100644 --- a/pkg/ethblocklistener/blocklistener_test.go +++ b/pkg/ethblocklistener/blocklistener_test.go @@ -55,6 +55,8 @@ func newTestBlockListener(t *testing.T, confSetup ...func(conf *BlockListenerCon MonitoredHeadLength: 50, HederaCompatibilityMode: false, BlockCacheSize: 250, + ReceiptCacheEnabled: true, + ReceiptCacheSize: 5000, } for _, fn := range confSetup { fn(conf, mRPC, cancelCtx) @@ -231,6 +233,14 @@ func TestBlockListenerConstructorFailCacheConfig(t *testing.T) { MonitoredHeadLength: 1, }, &ffresty.Config{}, &wsclient.WSConfig{}) require.Regexp(t, "FF23040", err) + + _, err = NewBlockListener(context.Background(), &retry.Retry{}, &BlockListenerConfig{ + BlockCacheSize: 250, + MonitoredHeadLength: 1, + ReceiptCacheEnabled: true, + ReceiptCacheSize: -1, + }, &ffresty.Config{}, &wsclient.WSConfig{}) + require.Regexp(t, "FF23040", err) } func TestBlockListenerStartGettingHighestBlockRetry(t *testing.T) { From 89529f6b55697f3843aa76d16601ba72c4ce9e52 Mon Sep 17 00:00:00 2001 From: Chengxuan Xing Date: Wed, 8 Jul 2026 13:20:41 +0100 Subject: [PATCH 2/2] adding comment for reseting cache Signed-off-by: Chengxuan Xing --- pkg/ethblocklistener/block_receipt_fetcher.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/pkg/ethblocklistener/block_receipt_fetcher.go b/pkg/ethblocklistener/block_receipt_fetcher.go index c51f6c9..49bcf86 100644 --- a/pkg/ethblocklistener/block_receipt_fetcher.go +++ b/pkg/ethblocklistener/block_receipt_fetcher.go @@ -105,6 +105,18 @@ func (brr *blockReceiptRequest) run() { earlyExit = false } +// 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. +// +// 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. func (bl *blockListener) resetReceiptCache() { if bl.txReceiptCache == nil { return