diff --git a/eth/api_backend.go b/eth/api_backend.go index 984e6ff5f6..766a99fc1e 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -36,12 +36,10 @@ import ( "github.com/ethereum/go-ethereum/core/txpool/locals" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" - "github.com/ethereum/go-ethereum/eth/filters" "github.com/ethereum/go-ethereum/eth/gasprice" "github.com/ethereum/go-ethereum/eth/tracers" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" - "github.com/ethereum/go-ethereum/internal/ethapi" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rpc" ) @@ -496,30 +494,3 @@ func (b *EthAPIBackend) RPCTxSyncDefaultTimeout() time.Duration { func (b *EthAPIBackend) RPCTxSyncMaxTimeout() time.Duration { return b.eth.config.TxSyncMaxTimeout } - -func (b *EthAPIBackend) SubscribeTransactionReceipts( - txHashes []common.Hash, - out chan<- []*ethapi.ReceiptWithTx, -) event.Subscription { - ch := make(chan core.ChainEvent, 16) - sub := b.eth.blockchain.SubscribeChainEvent(ch) - - go func() { - defer sub.Unsubscribe() - for { - select { - case ev, ok := <-ch: - if !ok { - return - } - batch := filters.FilterReceipts(txHashes, ev) - if len(batch) > 0 { - out <- batch - } - case <-sub.Err(): - return - } - } - }() - return sub -} diff --git a/eth/filters/filter.go b/eth/filters/filter.go index ee93a985f0..02399bc801 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -29,7 +29,6 @@ import ( "github.com/ethereum/go-ethereum/core/filtermaps" "github.com/ethereum/go-ethereum/core/history" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/internal/ethapi" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rpc" ) @@ -555,13 +554,16 @@ func bloomFilter(bloom types.Bloom, addresses []common.Address, topics [][]commo } // ReceiptWithTx contains a receipt and its corresponding transaction -type ReceiptWithTx = ethapi.ReceiptWithTx +type ReceiptWithTx struct { + Receipt *types.Receipt + Transaction *types.Transaction +} -// FilterReceipts returns the receipts matching the given criteria +// 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 { +func filterReceipts(txHashes []common.Hash, ev core.ChainEvent) []*ReceiptWithTx { var ret []*ReceiptWithTx receipts := ev.Receipts diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index 9210dba21c..02783fa5ec 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -445,7 +445,7 @@ func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent) // Handle transaction receipts subscriptions when a new block is added for _, f := range filters[TransactionReceiptsSubscription] { - matchedReceipts := FilterReceipts(f.txHashes, ev) + matchedReceipts := filterReceipts(f.txHashes, ev) if len(matchedReceipts) > 0 { f.receipts <- matchedReceipts } diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index a3d6aebd4d..bcd1fb7b2a 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -1652,11 +1652,6 @@ func (api *TransactionAPI) SendRawTransaction(ctx context.Context, input hexutil return SubmitTransaction(ctx, api.b, tx) } -type ReceiptWithTx struct { - Receipt *types.Receipt - Transaction *types.Transaction -} - // SendRawTransactionSync will add the signed transaction to the transaction pool // and wait until the transaction has been included in a block and return the receipt, or the timeout. func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hexutil.Bytes, timeoutMs *hexutil.Uint64) (map[string]interface{}, error) { @@ -1664,6 +1659,12 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex if err := tx.UnmarshalBinary(input); err != nil { return nil, err } + + ch := make(chan core.ChainEvent, 128) + sub := api.b.SubscribeChainEvent(ch) + subErrCh := sub.Err() + defer sub.Unsubscribe() + hash, err := SubmitTransaction(ctx, api.b, tx) if err != nil { return nil, err @@ -1690,13 +1691,6 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex return r, nil } - // Subscribe to receipt stream (filtered to this tx) - receipts := make(chan []*ReceiptWithTx, 1) - sub := api.b.SubscribeTransactionReceipts([]common.Hash{hash}, receipts) - defer sub.Unsubscribe() - - subErrCh := sub.Err() - for { select { case <-receiptCtx.Done(): @@ -1717,23 +1711,27 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex } return nil, err - case batch := <-receipts: - for _, rwt := range batch { - if rwt == nil || rwt.Receipt == nil || rwt.Receipt.TxHash != hash { - continue + case ev := <-ch: + rs := ev.Receipts + txs := ev.Transactions + if len(rs) == 0 || len(rs) != len(txs) { + continue + } + for i := range rs { + if rs[i].TxHash == hash { + if rs[i].BlockNumber != nil && rs[i].BlockHash != (common.Hash{}) { + signer := types.LatestSigner(api.b.ChainConfig()) + return MarshalReceipt( + rs[i], + rs[i].BlockHash, + rs[i].BlockNumber.Uint64(), + signer, + txs[i], + int(rs[i].TransactionIndex), + ), nil + } + return api.GetTransactionReceipt(receiptCtx, hash) } - - if rwt.Receipt.BlockNumber != nil && rwt.Receipt.BlockHash != (common.Hash{}) { - return MarshalReceipt( - rwt.Receipt, - rwt.Receipt.BlockHash, - rwt.Receipt.BlockNumber.Uint64(), - api.signer, - rwt.Transaction, - int(rwt.Receipt.TransactionIndex), - ), nil - } - return api.GetTransactionReceipt(receiptCtx, hash) } } } diff --git a/internal/ethapi/api_test.go b/internal/ethapi/api_test.go index c28b22e6ee..ad2af1ec11 100644 --- a/internal/ethapi/api_test.go +++ b/internal/ethapi/api_test.go @@ -441,8 +441,8 @@ type testBackend struct { pending *types.Block pendingReceipts types.Receipts - receiptsFeed *event.Feed - autoMine bool + chainFeed *event.Feed + autoMine bool sentTx *types.Transaction sentTxHash common.Hash @@ -479,7 +479,7 @@ func newTestBackend(t *testing.T, n int, gspec *core.Genesis, engine consensus.E acc: acc, pending: blocks[n], pendingReceipts: receipts[n], - receiptsFeed: new(event.Feed), + chainFeed: new(event.Feed), } return backend } @@ -601,7 +601,7 @@ func (b testBackend) GetEVM(ctx context.Context, state *state.StateDB, header *t return vm.NewEVM(context, state, b.chain.Config(), *vmConfig) } func (b testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription { - panic("implement me") + return b.chainFeed.Subscribe(ch) } func (b testBackend) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription { panic("implement me") @@ -610,11 +610,12 @@ func (b *testBackend) SendTx(ctx context.Context, tx *types.Transaction) error { b.sentTx = tx b.sentTxHash = tx.Hash() - if b.autoMine && b.receiptsFeed != nil { + if b.autoMine { + // Synthesize a "mined" receipt at head+1 num := b.chain.CurrentHeader().Number.Uint64() + 1 - bh := fakeBlockHash(tx.Hash()) - r := &types.Receipt{ + bh := fakeBlockHash(tx.Hash()) + receipt := &types.Receipt{ TxHash: tx.Hash(), Status: types.ReceiptStatusSuccessful, BlockHash: bh, @@ -623,7 +624,16 @@ func (b *testBackend) SendTx(ctx context.Context, tx *types.Transaction) error { CumulativeGasUsed: 21000, GasUsed: 21000, } - b.receiptsFeed.Send([]*ReceiptWithTx{{Receipt: r, Transaction: tx}}) + hdr := &types.Header{ + Number: new(big.Int).SetUint64(num), + } + + // Broadcast a ChainEvent that includes the receipts and txs + b.chainFeed.Send(core.ChainEvent{ + Header: hdr, + Receipts: types.Receipts{receipt}, + Transactions: types.Transactions{tx}, + }) } return nil } @@ -3981,13 +3991,6 @@ func makeSelfSignedRaw(t *testing.T, api *TransactionAPI, addr common.Address) ( return makeSignedRaw(t, api, addr, addr, big.NewInt(0)) } -func (b *testBackend) SubscribeTransactionReceipts(_ []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription { - if b.receiptsFeed == nil { - return event.NewSubscription(func(quit <-chan struct{}) error { <-quit; return nil }) - } - // Test will only publish the receipts it wants; we just forward. - return b.receiptsFeed.Subscribe(ch) -} func TestSendRawTransactionSync_Success(t *testing.T) { t.Parallel() genesis := &core.Genesis{ diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go index b67ecef322..af3d592b82 100644 --- a/internal/ethapi/backend.go +++ b/internal/ethapi/backend.go @@ -98,7 +98,6 @@ type Backend interface { GetLogs(ctx context.Context, blockHash common.Hash, number uint64) ([][]*types.Log, error) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription - SubscribeTransactionReceipts(txHashes []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription CurrentView() *filtermaps.ChainView NewMatcherBackend() filtermaps.MatcherBackend diff --git a/internal/ethapi/transaction_args_test.go b/internal/ethapi/transaction_args_test.go index dd75defecf..30791f32b5 100644 --- a/internal/ethapi/transaction_args_test.go +++ b/internal/ethapi/transaction_args_test.go @@ -411,6 +411,3 @@ func (b *backendMock) CurrentView() *filtermaps.ChainView { return nil func (b *backendMock) NewMatcherBackend() filtermaps.MatcherBackend { return nil } func (b *backendMock) HistoryPruningCutoff() uint64 { return 0 } -func (b *backendMock) SubscribeTransactionReceipts(txHashes []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription { - return nil -}