core, eth, internal: return tx indexing progress

This commit is contained in:
Gary Rong 2023-12-19 15:18:06 +08:00
parent 8b1664dac8
commit 076ba509be
4 changed files with 110 additions and 41 deletions

View file

@ -185,6 +185,29 @@ func DefaultCacheConfigWithScheme(scheme string) *CacheConfig {
return &config
}
// txLookup is wrapper over transaction lookup along with the corresponding
// transaction itself.
type txLookup struct {
lookup *rawdb.LegacyTxLookupEntry
transaction *types.Transaction
}
// txIndexProgress is the struct describing the progress for transaction indexing.
type txIndexProgress struct {
tail uint64 // the oldest block indexed for transactions
head uint64 // the latest block indexed for transactions
limit uint64 // the number of blocks required for transaction indexing(0 means the whole chain)
}
// Error implements Error returning the progress in string format.
func (prog txIndexProgress) Error() string {
limit := "entire chain"
if prog.limit != 0 {
limit = fmt.Sprintf("last %d blocks", prog.limit)
}
return fmt.Sprintf("index-tail: %d, index-head: %d, limit: %s", prog.tail, prog.head, limit)
}
// BlockChain represents the canonical chain given a database with a genesis
// block. The Blockchain manages chain imports, reverts, chain reorganisations.
//
@ -242,15 +265,16 @@ type BlockChain struct {
bodyRLPCache *lru.Cache[common.Hash, rlp.RawValue]
receiptsCache *lru.Cache[common.Hash, []*types.Receipt]
blockCache *lru.Cache[common.Hash, *types.Block]
txLookupCache *lru.Cache[common.Hash, *rawdb.LegacyTxLookupEntry]
txLookupCache *lru.Cache[common.Hash, txLookup]
// future blocks are blocks added for later processing
futureBlocks *lru.Cache[common.Hash, *types.Block]
wg sync.WaitGroup //
quit chan struct{} // shutdown signal, closed in Stop.
stopping atomic.Bool // false if chain is running, true when stopped
procInterrupt atomic.Bool // interrupt signaler for block processing
wg sync.WaitGroup
quit chan struct{} // shutdown signal, closed in Stop.
stopping atomic.Bool // false if chain is running, true when stopped
procInterrupt atomic.Bool // interrupt signaler for block processing
txIndexProgCh chan chan txIndexProgress // chan for querying the progress of transaction indexing
engine consensus.Engine
validator Validator // Block and state validator interface
@ -297,8 +321,9 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
bodyRLPCache: lru.NewCache[common.Hash, rlp.RawValue](bodyCacheLimit),
receiptsCache: lru.NewCache[common.Hash, []*types.Receipt](receiptsCacheLimit),
blockCache: lru.NewCache[common.Hash, *types.Block](blockCacheLimit),
txLookupCache: lru.NewCache[common.Hash, *rawdb.LegacyTxLookupEntry](txLookupCacheLimit),
txLookupCache: lru.NewCache[common.Hash, txLookup](txLookupCacheLimit),
futureBlocks: lru.NewCache[common.Hash, *types.Block](maxFutureBlocks),
txIndexProgCh: make(chan chan txIndexProgress),
engine: engine,
vmConfig: vmConfig,
}
@ -2425,6 +2450,34 @@ func (bc *BlockChain) indexBlocks(tail *uint64, head uint64, done chan struct{})
}
}
// reportTxIndexProgress returns the tx indexing progress.
func (bc *BlockChain) reportTxIndexProgress(head uint64) txIndexProgress {
tail := rawdb.ReadTxIndexTail(bc.db)
if tail == nil {
return txIndexProgress{
tail: 0, // not indexed yet
head: 0, // not indexed yet
limit: bc.txLookupLimit,
}
}
return txIndexProgress{
tail: *tail,
head: head,
limit: bc.txLookupLimit,
}
}
// askTxIndexProgress retrieves the tx indexing progress.
func (bc *BlockChain) askTxIndexProgress() (txIndexProgress, error) {
ch := make(chan txIndexProgress, 1)
select {
case bc.txIndexProgCh <- ch:
return <-ch, nil
case <-bc.quit:
return txIndexProgress{}, errors.New("blockchain is closed")
}
}
// maintainTxIndex is responsible for the construction and deletion of the
// transaction index.
//
@ -2440,8 +2493,9 @@ func (bc *BlockChain) maintainTxIndex() {
// Listening to chain events and manipulate the transaction indexes.
var (
done chan struct{} // Non-nil if background unindexing or reindexing routine is active.
headCh = make(chan ChainHeadEvent, 1) // Buffered to avoid locking up the event feed
done chan struct{} // Non-nil if background unindexing or reindexing routine is active.
lastHead uint64 // The latest announced chain head (whose tx indexes are assumed created)
headCh = make(chan ChainHeadEvent, 1) // Buffered to avoid locking up the event feed
)
sub := bc.SubscribeChainHeadEvent(headCh)
if sub == nil {
@ -2464,8 +2518,11 @@ func (bc *BlockChain) maintainTxIndex() {
done = make(chan struct{})
go bc.indexBlocks(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), done)
}
lastHead = head.Block.NumberU64()
case <-done:
done = nil
case ch := <-bc.txIndexProgCh:
ch <- bc.reportTxIndexProgress(lastHead)
case <-bc.quit:
if done != nil {
log.Info("Waiting background transaction indexer to exit")

View file

@ -254,20 +254,34 @@ func (bc *BlockChain) GetAncestor(hash common.Hash, number, ancestor uint64, max
return bc.hc.GetAncestor(hash, number, ancestor, maxNonCanonical)
}
// GetTransactionLookup retrieves the lookup associate with the given transaction
// hash from the cache or database.
func (bc *BlockChain) GetTransactionLookup(hash common.Hash) *rawdb.LegacyTxLookupEntry {
// GetTransactionLookup retrieves the lookup along with the transaction itself
// associate with the given transaction hash. A non-nil error will be returned
// if the transaction is not found.
func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLookupEntry, *types.Transaction, error) {
// Short circuit if the txlookup already in the cache, retrieve otherwise
if lookup, exist := bc.txLookupCache.Get(hash); exist {
return lookup
if item, exist := bc.txLookupCache.Get(hash); exist {
return item.lookup, item.transaction, nil
}
tx, blockHash, blockNumber, txIndex := rawdb.ReadTransaction(bc.db, hash)
if tx == nil {
return nil
// The transaction can either be non-existent, or just not indexed
// yet. Return the tx indexing progress as well for better UX.
progress, err := bc.askTxIndexProgress()
if err != nil {
return nil, nil, err
}
return nil, nil, progress
}
lookup := &rawdb.LegacyTxLookupEntry{BlockHash: blockHash, BlockIndex: blockNumber, Index: txIndex}
bc.txLookupCache.Add(hash, lookup)
return lookup
lookup := &rawdb.LegacyTxLookupEntry{
BlockHash: blockHash,
BlockIndex: blockNumber,
Index: txIndex,
}
bc.txLookupCache.Add(hash, txLookup{
lookup: lookup,
transaction: tx,
})
return lookup, tx, nil
}
// GetTd retrieves a block's total difficulty in the canonical chain from the

View file

@ -19,6 +19,7 @@ package eth
import (
"context"
"errors"
"fmt"
"math/big"
"time"
@ -308,9 +309,15 @@ func (b *EthAPIBackend) GetPoolTransaction(hash common.Hash) *types.Transaction
return b.eth.txPool.Get(hash)
}
// GetTransaction retrieves the lookup along with the transaction itself associate
// with the given transaction hash. A non-nil error will be returned if the
// transaction is not found.
func (b *EthAPIBackend) GetTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
tx, blockHash, blockNumber, index := rawdb.ReadTransaction(b.eth.ChainDb(), txHash)
return tx, blockHash, blockNumber, index, nil
lookup, tx, err := b.eth.blockchain.GetTransactionLookup(txHash)
if err != nil {
return nil, common.Hash{}, 0, 0, fmt.Errorf("tx is not existent or not indexed, %w", err)
}
return tx, lookup.BlockHash, lookup.BlockIndex, lookup.Index, nil
}
func (b *EthAPIBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {

View file

@ -1654,22 +1654,17 @@ func (s *TransactionAPI) GetTransactionByHash(ctx context.Context, hash common.H
// Try to return an already finalized transaction
tx, blockHash, blockNumber, index, err := s.b.GetTransaction(ctx, hash)
if err != nil {
// No finalized transaction, try to retrieve it from the pool
if tx := s.b.GetPoolTransaction(hash); tx != nil {
return NewRPCPendingTransaction(tx, s.b.CurrentHeader(), s.b.ChainConfig()), nil
}
return nil, err
}
if tx != nil {
header, err := s.b.HeaderByHash(ctx, blockHash)
if err != nil {
return nil, err
}
return newRPCTransaction(tx, blockHash, blockNumber, header.Time, index, header.BaseFee, s.b.ChainConfig()), nil
header, err := s.b.HeaderByHash(ctx, blockHash)
if err != nil {
return nil, err
}
// No finalized transaction, try to retrieve it from the pool
if tx := s.b.GetPoolTransaction(hash); tx != nil {
return NewRPCPendingTransaction(tx, s.b.CurrentHeader(), s.b.ChainConfig()), nil
}
// Transaction unknown, return as such
return nil, nil
return newRPCTransaction(tx, blockHash, blockNumber, header.Time, index, header.BaseFee, s.b.ChainConfig()), nil
}
// GetRawTransactionByHash returns the bytes of the transaction for the given hash.
@ -1677,12 +1672,8 @@ func (s *TransactionAPI) GetRawTransactionByHash(ctx context.Context, hash commo
// Retrieve a finalized transaction, or a pooled otherwise
tx, _, _, _, err := s.b.GetTransaction(ctx, hash)
if err != nil {
return nil, err
}
if tx == nil {
if tx = s.b.GetPoolTransaction(hash); tx == nil {
// Transaction not found anywhere, abort
return nil, nil
return nil, err
}
}
// Serialize to RLP and return
@ -1692,10 +1683,10 @@ func (s *TransactionAPI) GetRawTransactionByHash(ctx context.Context, hash commo
// GetTransactionReceipt returns the transaction receipt for the given transaction hash.
func (s *TransactionAPI) GetTransactionReceipt(ctx context.Context, hash common.Hash) (map[string]interface{}, error) {
tx, blockHash, blockNumber, index, err := s.b.GetTransaction(ctx, hash)
if tx == nil || err != nil {
// When the transaction doesn't exist, the RPC method should return JSON null
// as per specification.
return nil, nil
if err != nil {
// When the transaction doesn't exist or is not indexed yet,
// the RPC method should return JSON null as per specification.
return nil, err
}
header, err := s.b.HeaderByHash(ctx, blockHash)
if err != nil {