Adding an option to cache txn receipts when building canonical chain

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|

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")

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,

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)

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)

pkg/ethblocklistener/block_receipt_fetcher.go

@@ -104,3 +104,85 @@ func (brr *blockReceiptRequest) run() {
 	// No early return in this function - return must happen by reaching here
 	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
+	}
+	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))
+		}
+	}
+}

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 := &ethrpc.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) = &ethrpc.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)
+}

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"`
@@ -122,6 +124,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
 
@@ -133,6 +136,10 @@ type blockListener struct {
 	highestBlock        uint64
 	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
+
 	// headBlockNumber mode: last head value sent on the block listener channel (only written from listenLoop)
 	currentChainHead uint64
 }
@@ -178,6 +185,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
 }
 
@@ -521,7 +534,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 {
@@ -534,6 +549,7 @@ func (bl *blockListener) handleNewBlock(mbi *ethrpc.BlockInfoJSONRPC, addAfter *
 			toRemove := nextElem
 			nextElem = nextElem.Next()
 			_ = bl.canonicalChain.Remove(toRemove)
+			forkTrim = true
 		}
 	}
 
@@ -542,6 +558,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
@@ -553,8 +574,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 {
@@ -606,6 +629,7 @@ func (bl *blockListener) rebuildCanonicalChain() *list.Element {
 		}
 
 		bl.checkAndSetHighestBlock(bi)
+		bl.fetchAndCacheBlockReceipts(bi)
 
 	}
 	return notifyPos

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, &ethReceipt, "eth_getTransactionReceipt", txHash)
 	if rpcErr != nil || ethReceipt == nil {
 		var err error

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) {