websocket: optimize receipts retrieval

Author: 10gic

core/blockchain.go

@@ -1690,7 +1690,12 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
 	// Set new head.
 	bc.writeHeadBlock(block)
 
-	bc.chainFeed.Send(ChainEvent{Header: block.Header()})
+	bc.chainFeed.Send(ChainEvent{
+		Header:       block.Header(),
+		Receipts:     receipts,
+		Transactions: block.Transactions(),
+	})
+
 	if len(logs) > 0 {
 		bc.logsFeed.Send(logs)
 	}
@@ -2589,7 +2594,13 @@ func (bc *BlockChain) SetCanonical(head *types.Block) (common.Hash, error) {
 
 	// Emit events
 	logs := bc.collectLogs(head, false)
-	bc.chainFeed.Send(ChainEvent{Header: head.Header()})
+
+	bc.chainFeed.Send(ChainEvent{
+		Header:       head.Header(),
+		Receipts:     bc.GetReceiptsByHash(head.Hash()),
+		Transactions: head.Transactions(),
+	})
+
 	if len(logs) > 0 {
 		bc.logsFeed.Send(logs)
 	}

core/events.go

@@ -27,7 +27,9 @@ type NewTxsEvent struct{ Txs []*types.Transaction }
 type RemovedLogsEvent struct{ Logs []*types.Log }
 
 type ChainEvent struct {
-	Header *types.Header
+	Header       *types.Header
+	Receipts     []*types.Receipt
+	Transactions []*types.Transaction
 }
 
 type ChainHeadEvent struct {

eth/filters/api.go

@@ -318,16 +318,16 @@ func (api *FilterAPI) TransactionReceipts(ctx context.Context, filter *Transacti
 	}
 
 	var (
-		rpcSub           = notifier.CreateSubscription()
-		filteredReceipts = make(chan []*ReceiptWithTx)
-		txHashes         []common.Hash
+		rpcSub          = notifier.CreateSubscription()
+		matchedReceipts = make(chan []*ReceiptWithTx)
+		txHashes        []common.Hash
 	)
 
 	if filter != nil {
 		txHashes = filter.TransactionHashes
 	}
 
-	receiptsSub := api.events.SubscribeTransactionReceipts(txHashes, filteredReceipts)
+	receiptsSub := api.events.SubscribeTransactionReceipts(txHashes, matchedReceipts)
 
 	go func() {
 		defer receiptsSub.Unsubscribe()
@@ -336,11 +336,11 @@ func (api *FilterAPI) TransactionReceipts(ctx context.Context, filter *Transacti
 
 		for {
 			select {
-			case receiptsWithTx := <-filteredReceipts:
-				if len(receiptsWithTx) > 0 {
+			case receiptsWithTxs := <-matchedReceipts:
+				if len(receiptsWithTxs) > 0 {
 					// Convert to the same format as eth_getTransactionReceipt
-					marshaledReceipts := make([]map[string]interface{}, len(receiptsWithTx))
-					for i, receiptWithTx := range receiptsWithTx {
+					marshaledReceipts := make([]map[string]interface{}, len(receiptsWithTxs))
+					for i, receiptWithTx := range receiptsWithTxs {
 						marshaledReceipts[i] = ethapi.MarshalReceipt(
 							receiptWithTx.Receipt,
 							receiptWithTx.Receipt.BlockHash,

eth/filters/filter.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/filtermaps"
 	"github.com/ethereum/go-ethereum/core/history"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -551,3 +552,65 @@ func bloomFilter(bloom types.Bloom, addresses []common.Address, topics [][]commo
 	}
 	return true
 }
+
+// ReceiptWithTx contains a receipt and its corresponding transaction
+type ReceiptWithTx struct {
+	Receipt     *types.Receipt
+	Transaction *types.Transaction
+}
+
+// filterReceipts returns the receipts matching the given criteria
+// In addition to returning receipts, it also returns the corresponding transactions.
+// This is because receipts only contain low-level data, while user-facing data
+// may require additional information from the Transaction.
+func filterReceipts(txHashes []common.Hash, ev core.ChainEvent) []*ReceiptWithTx {
+	var ret []*ReceiptWithTx
+
+	receipts := ev.Receipts
+	txs := ev.Transactions
+
+	if len(txHashes) == 0 {
+		// No filter, send all receipts with their transactions.
+		ret = make([]*ReceiptWithTx, len(receipts))
+		for i, receipt := range receipts {
+			ret[i] = &ReceiptWithTx{
+				Receipt:     receipt,
+				Transaction: txs[i],
+			}
+		}
+	} else if len(txHashes) == 1 {
+		// Filter by single transaction hash.
+		// This is a common case, so we distinguish it from filtering by multiple tx hashes and made a small optimization.
+		for i, receipt := range receipts {
+			if receipt.TxHash == (txHashes)[0] {
+				ret = append(ret, &ReceiptWithTx{
+					Receipt:     receipt,
+					Transaction: txs[i],
+				})
+				break
+			}
+		}
+	} else {
+		// Filter by multiple transaction hashes.
+		txHashMap := make(map[common.Hash]bool, len(txHashes))
+		for _, hash := range txHashes {
+			txHashMap[hash] = true
+		}
+
+		for i, receipt := range receipts {
+			if txHashMap[receipt.TxHash] {
+				ret = append(ret, &ReceiptWithTx{
+					Receipt:     receipt,
+					Transaction: txs[i],
+				})
+
+				// Early exit if all receipts are found
+				if len(ret) == len(txHashes) {
+					break
+				}
+			}
+		}
+	}
+
+	return ret
+}

eth/filters/filter_system.go

@@ -444,92 +444,10 @@ func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent)
 	}
 
 	// Handle transaction receipts subscriptions when a new block is added
-	es.handleReceiptsEvent(filters, ev)
-}
-
-// ReceiptWithTx contains a receipt and its corresponding transaction for websocket subscription
-type ReceiptWithTx struct {
-	Receipt     *types.Receipt
-	Transaction *types.Transaction
-}
-
-func (es *EventSystem) handleReceiptsEvent(filters filterIndex, ev core.ChainEvent) {
-	// If there are no transaction receipt subscriptions, skip processing
-	if len(filters[TransactionReceiptsSubscription]) == 0 {
-		return
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	// Get receipts for this block
-	receipts, err := es.backend.GetReceipts(ctx, ev.Header.Hash())
-	if err != nil {
-		log.Warn("Failed to get receipts for block", "hash", ev.Header.Hash(), "err", err)
-		return
-	}
-
-	// Get body to retrieve transactions
-	body, err := es.backend.GetBody(ctx, ev.Header.Hash(), rpc.BlockNumber(ev.Header.Number.Int64()))
-	if err != nil {
-		log.Warn("Failed to get block for receipts", "hash", ev.Header.Hash(), "err", err)
-		return
-	}
-
-	txs := body.Transactions
-	if len(txs) != len(receipts) {
-		log.Warn("Transaction count mismatch", "txs", len(txs), "receipts", len(receipts))
-		return
-	}
-
 	for _, f := range filters[TransactionReceiptsSubscription] {
-		var filteredReceipts []*ReceiptWithTx
-
-		if len(f.txHashes) == 0 {
-			// No filter, send all receipts with their transactions.
-			filteredReceipts = make([]*ReceiptWithTx, len(receipts))
-			for i, receipt := range receipts {
-				filteredReceipts[i] = &ReceiptWithTx{
-					Receipt:     receipt,
-					Transaction: txs[i],
-				}
-			}
-		} else if len(f.txHashes) == 1 {
-			// Filter by single transaction hash.
-			// This is a common case, so we distinguish it from filtering by multiple tx hashes and made a small optimization.
-			for i, receipt := range receipts {
-				if receipt.TxHash == f.txHashes[0] {
-					filteredReceipts = append(filteredReceipts, &ReceiptWithTx{
-						Receipt:     receipt,
-						Transaction: txs[i],
-					})
-					break
-				}
-			}
-		} else {
-			// Filter by multiple transaction hashes.
-			txHashMap := make(map[common.Hash]bool, len(f.txHashes))
-			for _, hash := range f.txHashes {
-				txHashMap[hash] = true
-			}
-
-			for i, receipt := range receipts {
-				if txHashMap[receipt.TxHash] {
-					filteredReceipts = append(filteredReceipts, &ReceiptWithTx{
-						Receipt:     receipt,
-						Transaction: txs[i],
-					})
-
-					// Early exit if all receipts are found
-					if len(filteredReceipts) == len(f.txHashes) {
-						break
-					}
-				}
-			}
-		}
-
-		if len(filteredReceipts) > 0 {
-			f.receipts <- filteredReceipts
+		matchedReceipts := filterReceipts(f.txHashes, ev)
+		if len(matchedReceipts) > 0 {
+			f.receipts <- matchedReceipts
 		}
 	}
 }

eth/filters/filter_system_test.go

@@ -829,7 +829,17 @@ func TestTransactionReceiptsSubscription(t *testing.T) {
 	}
 
 	// Prepare test data
-	chainEvent := core.ChainEvent{Header: chain[0].Header()}
+	receipts := blockchain.GetReceiptsByHash(chain[0].Hash())
+	if receipts == nil {
+		t.Fatalf("failed to get receipts")
+	}
+
+	chainEvent := core.ChainEvent{
+		Header:       chain[0].Header(),
+		Receipts:     receipts,
+		Transactions: chain[0].Transactions(),
+	}
+
 	txHashes := make([]common.Hash, txNum)
 	for i := 0; i < txNum; i++ {
 		txHashes[i] = chain[0].Transactions()[i].Hash()