From 118155b5656be86b9e8c6069c169d7969dbd2f88 Mon Sep 17 00:00:00 2001 From: 10gic <2391796+10gic@users.noreply.github.com> Date: Sun, 21 Sep 2025 22:22:27 +0800 Subject: [PATCH] websocket: add `transactionReceipts` for receipts notification --- eth/filters/api.go | 68 ++++++++++++++++ eth/filters/filter_system.go | 117 ++++++++++++++++++++++++++ eth/filters/filter_system_test.go | 131 ++++++++++++++++++++++++++++++ internal/ethapi/api.go | 8 +- 4 files changed, 320 insertions(+), 4 deletions(-) diff --git a/eth/filters/api.go b/eth/filters/api.go index d678c40389..1aa8ff862a 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -43,6 +43,7 @@ var ( errPendingLogsUnsupported = errors.New("pending logs are not supported") errExceedMaxTopics = errors.New("exceed max topics") errExceedLogQueryLimit = errors.New("exceed max addresses or topics per search position") + errExceedMaxTxHashes = errors.New("exceed max number of transaction hashes allowed per transactionReceipts subscription") ) const ( @@ -50,6 +51,8 @@ const ( maxTopics = 4 // The maximum number of allowed topics within a topic criteria maxSubTopics = 1000 + // The maximum number of transaction hash criteria allowed in a single subscription + maxTxHashes = 200 ) // filter is a helper struct that holds meta information over the filter type @@ -295,6 +298,71 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc return rpcSub, nil } +// TransactionReceiptsFilter defines criteria for transaction receipts subscription. +// If TransactionHashes is nil or empty, receipts for all transactions included in new blocks will be delivered. +// Otherwise, only receipts for the specified transactions will be delivered. +type TransactionReceiptsFilter struct { + TransactionHashes []common.Hash `json:"transactionHashes,omitempty"` +} + +// TransactionReceipts creates a subscription that fires transaction receipts when transactions are included in blocks. +func (api *FilterAPI) TransactionReceipts(ctx context.Context, filter *TransactionReceiptsFilter) (*rpc.Subscription, error) { + notifier, supported := rpc.NotifierFromContext(ctx) + if !supported { + return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported + } + + // Validate transaction hashes limit + if filter != nil && len(filter.TransactionHashes) > maxTxHashes { + return nil, errExceedMaxTxHashes + } + + var ( + rpcSub = notifier.CreateSubscription() + filteredReceipts = make(chan []*ReceiptWithTx) + txHashes []common.Hash + ) + + if filter != nil { + txHashes = filter.TransactionHashes + } + + receiptsSub := api.events.SubscribeTransactionReceipts(txHashes, filteredReceipts) + + go func() { + defer receiptsSub.Unsubscribe() + + signer := types.LatestSigner(api.sys.backend.ChainConfig()) + + for { + select { + case receiptsWithTx := <-filteredReceipts: + if len(receiptsWithTx) > 0 { + // Convert to the same format as eth_getTransactionReceipt + marshaledReceipts := make([]map[string]interface{}, len(receiptsWithTx)) + for i, receiptWithTx := range receiptsWithTx { + marshaledReceipts[i] = ethapi.MarshalReceipt( + receiptWithTx.Receipt, + receiptWithTx.Receipt.BlockHash, + receiptWithTx.Receipt.BlockNumber.Uint64(), + signer, + receiptWithTx.Transaction, + int(receiptWithTx.Receipt.TransactionIndex), + ) + } + + // Send a batch of tx receipts in one notification + notifier.Notify(rpcSub.ID, marshaledReceipts) + } + case <-rpcSub.Err(): + return + } + } + }() + + return rpcSub, nil +} + // FilterCriteria represents a request to create a new filter. // Same as ethereum.FilterQuery but with UnmarshalJSON() method. type FilterCriteria ethereum.FilterQuery diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index ecf1c870c1..3520751634 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -158,6 +158,8 @@ const ( PendingTransactionsSubscription // BlocksSubscription queries hashes for blocks that are imported BlocksSubscription + // TransactionReceiptsSubscription queries for transaction receipts when transactions are included in blocks + TransactionReceiptsSubscription // LastIndexSubscription keeps track of the last index LastIndexSubscription ) @@ -182,6 +184,8 @@ type subscription struct { logs chan []*types.Log txs chan []*types.Transaction headers chan *types.Header + receipts chan []*ReceiptWithTx + txHashes []common.Hash // contains transaction hashes for transactionReceipts subscription filtering installed chan struct{} // closed when the filter is installed err chan error // closed when the filter is uninstalled } @@ -268,6 +272,7 @@ func (sub *Subscription) Unsubscribe() { case <-sub.f.logs: case <-sub.f.txs: case <-sub.f.headers: + case <-sub.f.receipts: } } @@ -353,6 +358,7 @@ func (es *EventSystem) subscribeLogs(crit ethereum.FilterQuery, logs chan []*typ logs: logs, txs: make(chan []*types.Transaction), headers: make(chan *types.Header), + receipts: make(chan []*ReceiptWithTx), installed: make(chan struct{}), err: make(chan error), } @@ -369,6 +375,7 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti logs: make(chan []*types.Log), txs: make(chan []*types.Transaction), headers: headers, + receipts: make(chan []*ReceiptWithTx), installed: make(chan struct{}), err: make(chan error), } @@ -385,6 +392,26 @@ func (es *EventSystem) SubscribePendingTxs(txs chan []*types.Transaction) *Subsc logs: make(chan []*types.Log), txs: txs, headers: make(chan *types.Header), + receipts: make(chan []*ReceiptWithTx), + installed: make(chan struct{}), + err: make(chan error), + } + return es.subscribe(sub) +} + +// SubscribeTransactionReceipts creates a subscription that writes transaction receipts for +// transactions when they are included in blocks. If txHashes is provided, only receipts +// for those specific transaction hashes will be delivered. +func (es *EventSystem) SubscribeTransactionReceipts(txHashes []common.Hash, receipts chan []*ReceiptWithTx) *Subscription { + sub := &subscription{ + id: rpc.NewID(), + typ: TransactionReceiptsSubscription, + created: time.Now(), + logs: make(chan []*types.Log), + txs: make(chan []*types.Transaction), + headers: make(chan *types.Header), + receipts: receipts, + txHashes: txHashes, installed: make(chan struct{}), err: make(chan error), } @@ -415,6 +442,96 @@ func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent) for _, f := range filters[BlocksSubscription] { f.headers <- ev.Header } + + // 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 + } + } } // eventLoop (un)installs filters and processes mux events. diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index 0048e74995..459cde8f79 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -31,6 +31,7 @@ import ( "github.com/ethereum/go-ethereum/core/filtermaps" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/internal/ethapi" @@ -781,3 +782,133 @@ func TestPendingTxFilterDeadlock(t *testing.T) { } } } + +// TestTransactionReceiptsSubscription tests the transaction receipts subscription functionality +func TestTransactionReceiptsSubscription(t *testing.T) { + t.Parallel() + + const txNum = 5 + + // Setup test environment + var ( + db = rawdb.NewMemoryDatabase() + backend, sys = newTestFilterSystem(db, Config{}) + api = NewFilterAPI(sys) + key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + addr1 = crypto.PubkeyToAddress(key1.PublicKey) + signer = types.NewLondonSigner(big.NewInt(1)) + genesis = &core.Genesis{ + Alloc: types.GenesisAlloc{addr1: {Balance: big.NewInt(1000000000000000000)}}, // 1 ETH + Config: params.TestChainConfig, + BaseFee: big.NewInt(params.InitialBaseFee), + } + _, chain, _ = core.GenerateChainWithGenesis(genesis, ethash.NewFaker(), 1, func(i int, gen *core.BlockGen) { + // Add transactions to the block + for j := 0; j < txNum; j++ { + toAddr := common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268") + tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{ + Nonce: uint64(j), + GasPrice: gen.BaseFee(), + Gas: 21000, + To: &toAddr, + Value: big.NewInt(1000), + Data: nil, + }), signer, key1) + gen.AddTx(tx) + } + }) + ) + + // Insert the blocks into the chain + blockchain, err := core.NewBlockChain(db, genesis, ethash.NewFaker(), nil) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + if n, err := blockchain.InsertChain(chain); err != nil { + t.Fatalf("block %d: failed to insert into chain: %v", n, err) + } + + // Prepare test data + chainEvent := core.ChainEvent{Header: chain[0].Header()} + txHashes := make([]common.Hash, txNum) + for i := 0; i < txNum; i++ { + txHashes[i] = chain[0].Transactions()[i].Hash() + } + + testCases := []struct { + name string + filterTxHashes []common.Hash + expectedReceiptTxHashes []common.Hash + expectError bool + }{ + { + name: "no filter - should return all receipts", + filterTxHashes: nil, + expectedReceiptTxHashes: txHashes, + expectError: false, + }, + { + name: "single tx hash filter", + filterTxHashes: []common.Hash{txHashes[0]}, + expectedReceiptTxHashes: []common.Hash{txHashes[0]}, + expectError: false, + }, + { + name: "multiple tx hashes filter", + filterTxHashes: []common.Hash{txHashes[0], txHashes[1], txHashes[2]}, + expectedReceiptTxHashes: []common.Hash{txHashes[0], txHashes[1], txHashes[2]}, + expectError: false, + }, + } + + // Run test cases + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + receiptsChan := make(chan []*ReceiptWithTx) + sub := api.events.SubscribeTransactionReceipts(tc.filterTxHashes, receiptsChan) + + // Send chain event + backend.chainFeed.Send(chainEvent) + + // Wait for receipts + timeout := time.After(1 * time.Second) + var receivedReceipts []*types.Receipt + for { + select { + case receiptsWithTx := <-receiptsChan: + for _, receiptWithTx := range receiptsWithTx { + receivedReceipts = append(receivedReceipts, receiptWithTx.Receipt) + } + case <-timeout: + t.Fatalf("timeout waiting for receipts") + } + if len(receivedReceipts) >= len(tc.expectedReceiptTxHashes) { + break + } + } + + // Verify receipt count + if len(receivedReceipts) != len(tc.expectedReceiptTxHashes) { + t.Errorf("Expected %d receipts, got %d", len(tc.expectedReceiptTxHashes), len(receivedReceipts)) + } + + // Verify specific transaction hashes are present + if tc.expectedReceiptTxHashes != nil { + receivedHashes := make(map[common.Hash]bool) + for _, receipt := range receivedReceipts { + receivedHashes[receipt.TxHash] = true + } + + for _, expectedHash := range tc.expectedReceiptTxHashes { + if !receivedHashes[expectedHash] { + t.Errorf("Expected receipt for tx %x not found", expectedHash) + } + } + } + + // Cleanup + sub.Unsubscribe() + <-sub.Err() + }) + } +} diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index ebb8ece730..83f8fa06c2 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -627,7 +627,7 @@ func (api *BlockChainAPI) GetBlockReceipts(ctx context.Context, blockNrOrHash rp result := make([]map[string]interface{}, len(receipts)) for i, receipt := range receipts { - result[i] = marshalReceipt(receipt, block.Hash(), block.NumberU64(), signer, txs[i], i) + result[i] = MarshalReceipt(receipt, block.Hash(), block.NumberU64(), signer, txs[i], i) } return result, nil } @@ -1472,11 +1472,11 @@ func (api *TransactionAPI) GetTransactionReceipt(ctx context.Context, hash commo return nil, err } // Derive the sender. - return marshalReceipt(receipt, blockHash, blockNumber, api.signer, tx, int(index)), nil + return MarshalReceipt(receipt, blockHash, blockNumber, api.signer, tx, int(index)), nil } -// marshalReceipt marshals a transaction receipt into a JSON object. -func marshalReceipt(receipt *types.Receipt, blockHash common.Hash, blockNumber uint64, signer types.Signer, tx *types.Transaction, txIndex int) map[string]interface{} { +// MarshalReceipt marshals a transaction receipt into a JSON object. +func MarshalReceipt(receipt *types.Receipt, blockHash common.Hash, blockNumber uint64, signer types.Signer, tx *types.Transaction, txIndex int) map[string]interface{} { from, _ := types.Sender(signer, tx) fields := map[string]interface{}{