use SubscribeTransactionReceipts

This commit is contained in:
aodhgan 2025-10-09 12:18:06 -07:00
parent a463b32a4b
commit 3a3c46a39b
4 changed files with 129 additions and 76 deletions

View file

@ -1652,6 +1652,11 @@ func (api *TransactionAPI) SendRawTransaction(ctx context.Context, input hexutil
return SubmitTransaction(ctx, api.b, tx) 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 // 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. // 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) { 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 return r, nil
} }
// Subscribe to new block events and check the receipt on each new block. // Subscribe to receipt stream (filtered to this tx)
heads := make(chan core.ChainHeadEvent, 16) rcpts := make(chan []*ReceiptWithTx, 1)
sub := api.b.SubscribeChainHeadEvent(heads) sub := api.b.SubscribeTransactionReceipts([]common.Hash{hash}, rcpts)
defer sub.Unsubscribe() defer sub.Unsubscribe()
subErrCh := sub.Err() 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 { for {
select { select {
case <-receiptCtx.Done(): case <-receiptCtx.Done():
// Upstream cancellation -> bubble it; otherwise emit our timeout error
if err := ctx.Err(); err != nil { if err := ctx.Err(); err != nil {
return nil, err return nil, err
} }
@ -1750,24 +1711,30 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex
case err, ok := <-subErrCh: case err, ok := <-subErrCh:
if !ok || err == nil { if !ok || err == nil {
// subscription closed; disable this case and rely on settle timer // subscription closed; disable this case
subErrCh = nil subErrCh = nil
continue continue
} }
return nil, err return nil, err
case <-heads: case batch := <-rcpts:
// Immediate re-check on new head, then grace to bridge indexing lag. for _, rwt := range batch {
if r, err := api.GetTransactionReceipt(receiptCtx, hash); err == nil && r != nil { if rwt == nil || rwt.Receipt == nil || rwt.Receipt.TxHash != hash {
return r, nil continue
} }
resetSettle(settleInterval)
case <-settleCh: if rwt.Receipt.BlockNumber != nil && rwt.Receipt.BlockHash != (common.Hash{}) {
if r, err := api.GetTransactionReceipt(receiptCtx, hash); err == nil && r != nil { return MarshalReceipt(
return r, nil 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)
} }
} }
} }

View file

@ -442,13 +442,19 @@ type testBackend struct {
pendingReceipts types.Receipts pendingReceipts types.Receipts
// test-only fields for SendRawTransactionSync // test-only fields for SendRawTransactionSync
autoMine bool receiptsFeed *event.Feed
mined bool headFeed *event.Feed // keep if other tests use it; otherwise remove
autoMine bool
sentTx *types.Transaction
sentTxHash common.Hash
syncDefaultTO time.Duration syncDefaultTO time.Duration
syncMaxTO 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 { 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], pending: blocks[n],
pendingReceipts: receipts[n], pendingReceipts: receipts[n],
headFeed: new(event.Feed), headFeed: new(event.Feed),
receiptsFeed: new(event.Feed),
} }
return backend 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 { func (b *testBackend) SendTx(ctx context.Context, tx *types.Transaction) error {
b.sentTx = tx b.sentTx = tx
b.sentTxHash = tx.Hash() 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 return nil
} }
func (b testBackend) GetCanonicalTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) { func (b *testBackend) GetCanonicalTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) {
// in-memory fast path for tests // Treat the auto-mined tx as canonically placed at head+1.
if b.mined && txHash == b.sentTxHash { if b.autoMine && txHash == b.sentTxHash {
return true, b.sentTx, common.HexToHash("0x01"), 1, 0 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) tx, blockHash, blockNumber, index := rawdb.ReadCanonicalTransaction(b.db, txHash)
return tx != nil, tx, blockHash, blockNumber, index return tx != nil, tx, blockHash, blockNumber, index
} }
func (b testBackend) GetCanonicalReceipt(tx *types.Transaction, blockHash common.Hash, blockNumber, blockIndex uint64) (*types.Receipt, error) { 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.autoMine && tx != nil && tx.Hash() == b.sentTxHash &&
if b.mined && tx != nil && tx.Hash() == b.sentTxHash && blockHash == common.HexToHash("0x01") && blockNumber == 1 && blockIndex == 0 { blockHash == fakeBlockHash(tx.Hash()) &&
blockIndex == 0 &&
blockNumber == b.chain.CurrentHeader().Number.Uint64()+1 {
return &types.Receipt{ return &types.Receipt{
Type: tx.Type(), Type: tx.Type(),
Status: types.ReceiptStatusSuccessful, Status: types.ReceiptStatusSuccessful,
@ -631,9 +653,9 @@ func (b testBackend) GetCanonicalReceipt(tx *types.Transaction, blockHash common
BlockHash: blockHash, BlockHash: blockHash,
BlockNumber: new(big.Int).SetUint64(blockNumber), BlockNumber: new(big.Int).SetUint64(blockNumber),
TransactionIndex: 0, TransactionIndex: 0,
TxHash: tx.Hash(),
}, nil }, nil
} }
// Fallback: use the chain's canonical receipt (DB-backed)
return b.chain.GetCanonicalReceipt(tx, blockHash, blockNumber, blockIndex) return b.chain.GetCanonicalReceipt(tx, blockHash, blockNumber, blockIndex)
} }
func (b testBackend) TxIndexDone() bool { 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) { func makeSelfSignedRaw(t *testing.T, api *TransactionAPI, addr common.Address) (hexutil.Bytes, *types.Transaction) {
return makeSignedRaw(t, api, addr, addr, big.NewInt(0)) 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) { func TestSendRawTransactionSync_Success(t *testing.T) {
t.Parallel() t.Parallel()
genesis := &core.Genesis{ genesis := &core.Genesis{

View file

@ -98,6 +98,7 @@ type Backend interface {
GetLogs(ctx context.Context, blockHash common.Hash, number uint64) ([][]*types.Log, error) GetLogs(ctx context.Context, blockHash common.Hash, number uint64) ([][]*types.Log, error)
SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
SubscribeTransactionReceipts(txHashes []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription
CurrentView() *filtermaps.ChainView CurrentView() *filtermaps.ChainView
NewMatcherBackend() filtermaps.MatcherBackend NewMatcherBackend() filtermaps.MatcherBackend

View file

@ -411,3 +411,6 @@ func (b *backendMock) CurrentView() *filtermaps.ChainView { return nil
func (b *backendMock) NewMatcherBackend() filtermaps.MatcherBackend { return nil } func (b *backendMock) NewMatcherBackend() filtermaps.MatcherBackend { return nil }
func (b *backendMock) HistoryPruningCutoff() uint64 { return 0 } func (b *backendMock) HistoryPruningCutoff() uint64 { return 0 }
func (b *backendMock) SubscribeTransactionReceipts(txHashes []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription {
return nil
}