forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 05:06:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

websocket: optimize receipts retrieval

Browse source
This commit is contained in:
10gic 2025-09-25 02:13:21 +08:00
parent 118155b565
commit 148d074a36
6 changed files with 101 additions and 97 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
core/blockchain.go
View file

@ -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)
}

4
core/events.go
View file

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

16
eth/filters/api.go
View file

@ -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,

63
eth/filters/filter.go
View file

@ -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
}

88
eth/filters/filter_system.go
View file

@ -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
}
}
}

12
eth/filters/filter_system_test.go
View file

@ -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()