diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index 1e3dc657a7..ccc0c3c7a7 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -1652,6 +1652,11 @@ func (api *TransactionAPI) SendRawTransaction(ctx context.Context, input hexutil return SubmitTransaction(ctx, api.b, tx) } +type ReceiptWithTx struct { + Receipt *types.Receipt + Tx *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) { @@ -1685,61 +1690,17 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex return r, nil } - // Subscribe to new block events and check the receipt on each new block. - heads := make(chan core.ChainHeadEvent, 16) - sub := api.b.SubscribeChainHeadEvent(heads) + // Subscribe to receipt stream (filtered to this tx) + rcpts := make(chan []*ReceiptWithTx, 1) + sub := api.b.SubscribeTransactionReceipts([]common.Hash{hash}, rcpts) defer sub.Unsubscribe() + subErrCh := sub.Err() - // calculate poll/settle intervals - const ( - pollFraction = 20 - pollMin = 25 * time.Millisecond - pollMax = 500 * time.Millisecond - ) - settleInterval := timeout / pollFraction - if settleInterval < pollMin { - settleInterval = pollMin - } else if settleInterval > pollMax { - settleInterval = pollMax - } - - // Settle: short delay to bridge receipt-indexing lag after a new head. - // resetSettle re-arms a single timer safely (stop+drain+reset). - // On head: check once immediately, then reset; on timer: re-check; repeat until deadline. - var ( - settle *time.Timer - settleCh <-chan time.Time - ) - resetSettle := func(d time.Duration) { - if settle == nil { - settle = time.NewTimer(d) - settleCh = settle.C - return - } - if !settle.Stop() { - select { - case <-settle.C: - default: - } - } - settle.Reset(d) - } - defer func() { - if settle != nil && !settle.Stop() { - select { - case <-settle.C: - default: - } - } - }() - - // Start a settle cycle immediately to cover a missed-head race. - resetSettle(settleInterval) - for { select { case <-receiptCtx.Done(): + // Upstream cancellation -> bubble it; otherwise emit our timeout error if err := ctx.Err(); err != nil { return nil, err } @@ -1750,24 +1711,30 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex case err, ok := <-subErrCh: if !ok || err == nil { - // subscription closed; disable this case and rely on settle timer + // subscription closed; disable this case subErrCh = nil continue } return nil, err - case <-heads: - // Immediate re-check on new head, then grace to bridge indexing lag. - if r, err := api.GetTransactionReceipt(receiptCtx, hash); err == nil && r != nil { - return r, nil - } - resetSettle(settleInterval) + case batch := <-rcpts: + for _, rwt := range batch { + if rwt == nil || rwt.Receipt == nil || rwt.Receipt.TxHash != hash { + continue + } - case <-settleCh: - if r, err := api.GetTransactionReceipt(receiptCtx, hash); err == nil && r != nil { - return r, nil + if rwt.Receipt.BlockNumber != nil && rwt.Receipt.BlockHash != (common.Hash{}) { + return MarshalReceipt( + rwt.Receipt, + rwt.Receipt.BlockHash, + rwt.Receipt.BlockNumber.Uint64(), + api.signer, + rwt.Tx, + int(rwt.Receipt.TransactionIndex), + ), nil + } + return api.GetTransactionReceipt(receiptCtx, hash) } - resetSettle(settleInterval) } } } diff --git a/internal/ethapi/api_test.go b/internal/ethapi/api_test.go index 5ee3453a1a..94d9982a83 100644 --- a/internal/ethapi/api_test.go +++ b/internal/ethapi/api_test.go @@ -442,13 +442,19 @@ type testBackend struct { pendingReceipts types.Receipts // test-only fields for SendRawTransactionSync - autoMine bool - mined bool + receiptsFeed *event.Feed + headFeed *event.Feed // keep if other tests use it; otherwise remove + autoMine bool + + sentTx *types.Transaction + sentTxHash common.Hash + syncDefaultTO time.Duration syncMaxTO time.Duration - sentTx *types.Transaction - sentTxHash common.Hash - headFeed *event.Feed +} + +func fakeBlockHash(txh common.Hash) common.Hash { + return crypto.Keccak256Hash([]byte("testblock"), txh.Bytes()) } func newTestBackend(t *testing.T, n int, gspec *core.Genesis, engine consensus.Engine, generator func(i int, b *core.BlockGen)) *testBackend { @@ -476,6 +482,7 @@ func newTestBackend(t *testing.T, n int, gspec *core.Genesis, engine consensus.E pending: blocks[n], pendingReceipts: receipts[n], headFeed: new(event.Feed), + receiptsFeed: new(event.Feed), } return backend } @@ -605,23 +612,38 @@ func (b testBackend) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) even func (b *testBackend) SendTx(ctx context.Context, tx *types.Transaction) error { b.sentTx = tx b.sentTxHash = tx.Hash() - if b.autoMine { - b.mined = true + + if b.autoMine && b.receiptsFeed != nil { + num := b.chain.CurrentHeader().Number.Uint64() + 1 + bh := fakeBlockHash(tx.Hash()) + + r := &types.Receipt{ + TxHash: tx.Hash(), + Status: types.ReceiptStatusSuccessful, + BlockHash: bh, + BlockNumber: new(big.Int).SetUint64(num), + TransactionIndex: 0, + CumulativeGasUsed: 21000, + GasUsed: 21000, + } + b.receiptsFeed.Send([]*ReceiptWithTx{{Receipt: r, Tx: tx}}) } return nil } -func (b testBackend) GetCanonicalTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) { - // in-memory fast path for tests - if b.mined && txHash == b.sentTxHash { - return true, b.sentTx, common.HexToHash("0x01"), 1, 0 +func (b *testBackend) GetCanonicalTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) { + // Treat the auto-mined tx as canonically placed at head+1. + if b.autoMine && txHash == b.sentTxHash { + num := b.chain.CurrentHeader().Number.Uint64() + 1 + return true, b.sentTx, fakeBlockHash(txHash), num, 0 } - // fallback to existing DB-backed path tx, blockHash, blockNumber, index := rawdb.ReadCanonicalTransaction(b.db, txHash) return tx != nil, tx, blockHash, blockNumber, index } -func (b testBackend) GetCanonicalReceipt(tx *types.Transaction, blockHash common.Hash, blockNumber, blockIndex uint64) (*types.Receipt, error) { - // In-memory fast path used by tests - if b.mined && tx != nil && tx.Hash() == b.sentTxHash && blockHash == common.HexToHash("0x01") && blockNumber == 1 && blockIndex == 0 { +func (b *testBackend) GetCanonicalReceipt(tx *types.Transaction, blockHash common.Hash, blockNumber, blockIndex uint64) (*types.Receipt, error) { + if b.autoMine && tx != nil && tx.Hash() == b.sentTxHash && + blockHash == fakeBlockHash(tx.Hash()) && + blockIndex == 0 && + blockNumber == b.chain.CurrentHeader().Number.Uint64()+1 { return &types.Receipt{ Type: tx.Type(), Status: types.ReceiptStatusSuccessful, @@ -631,9 +653,9 @@ func (b testBackend) GetCanonicalReceipt(tx *types.Transaction, blockHash common BlockHash: blockHash, BlockNumber: new(big.Int).SetUint64(blockNumber), TransactionIndex: 0, + TxHash: tx.Hash(), }, nil } - // Fallback: use the chain's canonical receipt (DB-backed) return b.chain.GetCanonicalReceipt(tx, blockHash, blockNumber, blockIndex) } func (b testBackend) TxIndexDone() bool { @@ -3961,6 +3983,66 @@ func makeSignedRaw(t *testing.T, api *TransactionAPI, from, to common.Address, v func makeSelfSignedRaw(t *testing.T, api *TransactionAPI, addr common.Address) (hexutil.Bytes, *types.Transaction) { return makeSignedRaw(t, api, addr, addr, big.NewInt(0)) } + +func (b *testBackend) SubscribeTransactionReceipts(txHashes []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription { + // If no feed is wired for this test, return a no-op subscription + if b.receiptsFeed == nil { + return event.NewSubscription(func(quit <-chan struct{}) error { + <-quit + return nil + }) + } + + // No filter => forward batches directly + if len(txHashes) == 0 { + return b.receiptsFeed.Subscribe(ch) + } + + // Filtered: wrap the underlying feed and only forward matching receipts + in := make(chan []*ReceiptWithTx, 1) + sub := b.receiptsFeed.Subscribe(in) + + // Build a hash set for quick filtering + wanted := make(map[common.Hash]struct{}, len(txHashes)) + for _, h := range txHashes { + wanted[h] = struct{}{} + } + + return event.NewSubscription(func(quit <-chan struct{}) error { + defer sub.Unsubscribe() + for { + select { + case batch, ok := <-in: + if !ok { + return nil + } + var out []*ReceiptWithTx + for _, r := range batch { + if r != nil && r.Receipt != nil { + if _, ok := wanted[r.Receipt.TxHash]; ok { + out = append(out, r) + } + } + } + if len(out) == 0 { + continue + } + select { + case ch <- out: + case <-quit: + return nil + } + case err, ok := <-sub.Err(): + if !ok || err == nil { + return nil + } + return err + case <-quit: + return nil + } + } + }) +} func TestSendRawTransactionSync_Success(t *testing.T) { t.Parallel() genesis := &core.Genesis{ diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go index af3d592b82..b67ecef322 100644 --- a/internal/ethapi/backend.go +++ b/internal/ethapi/backend.go @@ -98,6 +98,7 @@ 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 30791f32b5..dd75defecf 100644 --- a/internal/ethapi/transaction_args_test.go +++ b/internal/ethapi/transaction_args_test.go @@ -411,3 +411,6 @@ 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 +}