Merge pull request #2353 from CortexFoundation/dev

reduce api load

Author: ucwong

core/blockchain_reader.go

@@ -272,36 +272,18 @@ func (bc *BlockChain) GetAncestor(hash common.Hash, number, ancestor uint64, max
 // A null will be returned in the transaction is not found and background
 // transaction indexing is already finished. The transaction is not existent
 // from the node's perspective
-func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLookupEntry, *types.Transaction, error) {
+func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLookupEntry, *types.Transaction) {
 	bc.txLookupLock.RLock()
 	defer bc.txLookupLock.RUnlock()
 
 	// Short circuit if the txlookup already in the cache, retrieve otherwise
 	if item, exist := bc.txLookupCache.Get(hash); exist {
-		return item.lookup, item.transaction, nil
+		return item.lookup, item.transaction
 	}
 
 	tx, blockHash, blockNumber, txIndex := rawdb.ReadTransaction(bc.db, hash)
 	if tx == nil {
-		progress, err := bc.TxIndexProgress()
-		if err != nil {
-			// No error is returned if the transaction indexing progress is unreachable
-			// due to unexpected internal errors. In such cases, it is impossible to
-			// determine whether the transaction does not exist or has simply not been
-			// indexed yet without a progress marker.
-			//
-			// In such scenarios, the transaction is treated as unreachable, though
-			// this is clearly an unintended and unexpected situation.
-			return nil, nil, nil
-		}
-		// The transaction indexing is not finished yet, returning an
-		// error to explicitly indicate it.
-		if !progress.Done() {
-			return nil, nil, errors.New("transaction indexing still in progress")
-		}
-		// The transaction is already indexed, the transaction is either
-		// not existent or not in the range of index, returning null.
-		return nil, nil, nil
+		return nil, nil
 	}
 
 	lookup := &rawdb.LegacyTxLookupEntry{
@@ -313,7 +295,23 @@ func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLoo
 		lookup:      lookup,
 		transaction: tx,
 	})
-	return lookup, tx, nil
+	return lookup, tx
+}
+
+// TxIndexDone returns true if the transaction indexer has finished indexing.
+func (bc *BlockChain) TxIndexDone() bool {
+	progress, err := bc.TxIndexProgress()
+	if err != nil {
+		// No error is returned if the transaction indexing progress is unreachable
+		// due to unexpected internal errors. In such cases, it is impossible to
+		// determine whether the transaction does not exist or has simply not been
+		// indexed yet without a progress marker.
+		//
+		// In such scenarios, the transaction is treated as unreachable, though
+		// this is clearly an unintended and unexpected situation.
+		return true
+	}
+	return progress.Done()
 }
 
 // GetTd retrieves a block's total difficulty in the canonical chain from the
@@ -416,7 +414,7 @@ func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) {
 	if bc.txIndexer == nil {
 		return TxIndexProgress{}, errors.New("tx indexer is not enabled")
 	}
-	return bc.txIndexer.txIndexProgress()
+	return bc.txIndexer.txIndexProgress(), nil
 }
 
 // GetVMConfig returns the block chain VM config.

core/txindexer.go

@@ -17,8 +17,8 @@
 package core
 
 import (
-	"errors"
 	"fmt"
+	"sync/atomic"
 
 	"github.com/CortexFoundation/CortexTheseus/common"
 	"github.com/CortexFoundation/CortexTheseus/core/rawdb"
@@ -46,27 +46,38 @@ type txIndexer struct {
 	//  * N: means the latest N blocks [HEAD-N+1, HEAD] should be indexed
 	//       and all others shouldn't.
 	limit uint64
+	// The current head of blockchain for transaction indexing. This field
+	// is accessed by both the indexer and the indexing progress queries.
+	head atomic.Uint64
+
+	// The current tail of the indexed transactions, null indicates
+	// that no transactions have been indexed yet.
+	//
+	// This field is accessed by both the indexer and the indexing
+	// progress queries.
+	tail atomic.Pointer[uint64]
 
 	// cutoff denotes the block number before which the chain segment should
 	// be pruned and not available locally.
-	cutoff   uint64
-	db       ctxcdb.Database
-	progress chan chan TxIndexProgress
-	term     chan chan struct{}
-	closed   chan struct{}
+	cutoff uint64
+	db     ctxcdb.Database
+	term   chan chan struct{}
+	closed chan struct{}
 }
 
 // newTxIndexer initializes the transaction indexer.
 func newTxIndexer(limit uint64, chain *BlockChain) *txIndexer {
 	cutoff, _ := chain.HistoryPruningCutoff()
 	indexer := &txIndexer{
-		limit:    limit,
-		cutoff:   cutoff,
-		db:       chain.db,
-		progress: make(chan chan TxIndexProgress),
-		term:     make(chan chan struct{}),
-		closed:   make(chan struct{}),
+		limit:  limit,
+		cutoff: cutoff,
+		db:     chain.db,
+		term:   make(chan chan struct{}),
+		closed: make(chan struct{}),
 	}
+	indexer.head.Store(indexer.resolveHead())
+	indexer.tail.Store(rawdb.ReadTxIndexTail(chain.db))
+
 	go indexer.loop(chain)
 
 	var msg string
@@ -154,6 +165,7 @@ func (indexer *txIndexer) repair(head uint64) {
 		// A crash may occur between the two delete operations,
 		// potentially leaving dangling indexes in the database.
 		// However, this is considered acceptable.
+		indexer.tail.Store(nil)
 		rawdb.DeleteTxIndexTail(indexer.db)
 		rawdb.DeleteAllTxLookupEntries(indexer.db, nil)
 		log.Warn("Purge transaction indexes", "head", head, "tail", *tail)
@@ -174,6 +186,7 @@ func (indexer *txIndexer) repair(head uint64) {
 		// Traversing the database directly within the transaction
 		// index namespace might be slow and expensive, but we
 		// have no choice.
+		indexer.tail.Store(nil)
 		rawdb.DeleteTxIndexTail(indexer.db)
 		rawdb.DeleteAllTxLookupEntries(indexer.db, nil)
 		log.Warn("Purge transaction indexes", "head", head, "cutoff", indexer.cutoff)
@@ -187,6 +200,7 @@ func (indexer *txIndexer) repair(head uint64) {
 		// A crash may occur between the two delete operations,
 		// potentially leaving dangling indexes in the database.
 		// However, this is considered acceptable.
+		indexer.tail.Store(&indexer.cutoff)
 		rawdb.WriteTxIndexTail(indexer.db, indexer.cutoff)
 		rawdb.DeleteAllTxLookupEntries(indexer.db, func(txhash common.Hash, blob []byte) bool {
 			n := rawdb.DecodeTxLookupEntry(blob, indexer.db)
@@ -216,16 +230,16 @@ func (indexer *txIndexer) loop(chain *BlockChain) {
 
 	// Listening to chain events and manipulate the transaction indexes.
 	var (
-		stop chan struct{}           // Non-nil if background routine is active
-		done chan struct{}           // Non-nil if background routine is active
-		head = indexer.resolveHead() // The latest announced chain head
+		stop chan struct{} // Non-nil if background routine is active
+		done chan struct{} // Non-nil if background routine is active
 
 		headCh = make(chan ChainHeadEvent)
 		sub    = chain.SubscribeChainHeadEvent(headCh)
 	)
 	defer sub.Unsubscribe()
 
 	// Validate the transaction indexes and repair if necessary
+	head := indexer.head.Load()
 	indexer.repair(head)
 
 	// Launch the initial processing if chain is not empty (head != genesis).
@@ -238,17 +252,16 @@ func (indexer *txIndexer) loop(chain *BlockChain) {
 	for {
 		select {
 		case h := <-headCh:
+			indexer.head.Store(h.Header.Number.Uint64())
 			if done == nil {
 				stop = make(chan struct{})
 				done = make(chan struct{})
 				go indexer.run(h.Header.Number.Uint64(), stop, done)
 			}
-			head = h.Header.Number.Uint64()
 		case <-done:
 			stop = nil
 			done = nil
-		case ch := <-indexer.progress:
-			ch <- indexer.report(head)
+			indexer.tail.Store(rawdb.ReadTxIndexTail(indexer.db))
 		case ch := <-indexer.term:
 			if stop != nil {
 				close(stop)
@@ -264,7 +277,7 @@ func (indexer *txIndexer) loop(chain *BlockChain) {
 }
 
 // report returns the tx indexing progress.
-func (indexer *txIndexer) report(head uint64) TxIndexProgress {
+func (indexer *txIndexer) report(head uint64, tail *uint64) TxIndexProgress {
 	// Special case if the head is even below the cutoff,
 	// nothing to index.
 	if head < indexer.cutoff {
@@ -284,7 +297,6 @@ func (indexer *txIndexer) report(head uint64) TxIndexProgress {
 	}
 	// Compute how many blocks have been indexed
 	var indexed uint64
-	tail := rawdb.ReadTxIndexTail(indexer.db)
 	if tail != nil {
 		indexed = head - *tail + 1
 	}
@@ -302,14 +314,8 @@ func (indexer *txIndexer) report(head uint64) TxIndexProgress {
 
 // txIndexProgress retrieves the tx indexing progress, or an error if the
 // background tx indexer is already stopped.
-func (indexer *txIndexer) txIndexProgress() (TxIndexProgress, error) {
-	ch := make(chan TxIndexProgress, 1)
-	select {
-	case indexer.progress <- ch:
-		return <-ch, nil
-	case <-indexer.closed:
-		return TxIndexProgress{}, errors.New("indexer is closed")
-	}
+func (indexer *txIndexer) txIndexProgress() TxIndexProgress {
+	return indexer.report(indexer.head.Load(), indexer.tail.Load())
 }
 
 // close shutdown the indexer. Safe to be called for multiple times.

core/types/transaction.go

@@ -142,6 +142,13 @@ func (tx *Transaction) EncodeRLP(w io.Writer) error {
 	return rlp.Encode(w, &tx.data)
 }
 
+// MarshalBinary returns the canonical encoding of the transaction.
+// For legacy transactions, it returns the RLP encoding. For EIP-2718 typed
+// transactions, it returns the type and payload.
+func (tx *Transaction) MarshalBinary() ([]byte, error) {
+	return rlp.EncodeToBytes(tx)
+}
+
 // DecodeRLP implements rlp.Decoder
 func (tx *Transaction) DecodeRLP(s *rlp.Stream) error {
 	_, size, _ := s.Kind()

ctxc/api_backend.go

@@ -278,15 +278,17 @@ func (b *CortexAPIBackend) GetPoolTransaction(hash common.Hash) *types.Transacti
 	return b.ctxc.txPool.Get(hash)
 }
 
-func (b *CortexAPIBackend) GetTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
-	lookup, tx, err := b.ctxc.blockchain.GetTransactionLookup(txHash)
-	if err != nil {
-		return nil, common.Hash{}, 0, 0, err
-	}
+func (b *CortexAPIBackend) GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) {
+	lookup, tx := b.ctxc.blockchain.GetTransactionLookup(txHash)
 	if lookup == nil || tx == nil {
-		return nil, common.Hash{}, 0, 0, nil
+		return false, nil, common.Hash{}, 0, 0
 	}
-	return tx, lookup.BlockHash, lookup.BlockIndex, lookup.Index, nil
+	return true, tx, lookup.BlockHash, lookup.BlockIndex, lookup.Index
+}
+
+// TxIndexDone returns true if the transaction indexer has finished indexing.
+func (b *CortexAPIBackend) TxIndexDone() bool {
+	return b.ctxc.blockchain.TxIndexDone()
 }
 
 func (b *CortexAPIBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {
@@ -313,8 +315,13 @@ func (b *CortexAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) even
 	return b.ctxc.TxPool().SubscribeNewTxsEvent(ch)
 }
 
-func (b *CortexAPIBackend) SyncProgress() cortex.SyncProgress {
-	return b.ctxc.Downloader().Progress()
+func (b *CortexAPIBackend) SyncProgress(ctx context.Context) cortex.SyncProgress {
+	prog := b.ctxc.Downloader().Progress()
+	if txProg, err := b.ctxc.blockchain.TxIndexProgress(); err == nil {
+		prog.TxIndexFinishedBlocks = txProg.Indexed
+		prog.TxIndexRemainingBlocks = txProg.Remaining
+	}
+	return prog
 }
 
 func (b *CortexAPIBackend) ProtocolVersion() int {

ctxc/tracers/api.go

@@ -75,7 +75,8 @@ type Backend interface {
 	HeaderByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Header, error)
 	BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)
 	BlockByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Block, error)
-	GetTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error)
+	GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64)
+	TxIndexDone() bool
 	RPCGasCap() uint64
 	ChainConfig() *params.ChainConfig
 	Engine() consensus.Engine
@@ -807,9 +808,13 @@ func containsTx(block *types.Block, hash common.Hash) bool {
 // TraceTransaction returns the structured logs created during the execution of CVM
 // and returns them as a JSON object.
 func (api *API) TraceTransaction(ctx context.Context, hash common.Hash, config *TraceConfig) (any, error) {
-	tx, blockHash, blockNumber, index, err := api.backend.GetTransaction(ctx, hash)
-	if err != nil {
-		return nil, err
+	found, tx, blockHash, blockNumber, index := api.backend.GetTransaction(hash)
+	if !found {
+		// Warn in case tx indexer is not done.
+		if !api.backend.TxIndexDone() {
+			return nil, errTxNotFound
+		}
+		return nil, errTxNotFound
 	}
 
 	// Only mined txes are supported

interfaces.go

@@ -104,6 +104,10 @@ type SyncProgress struct {
 	HighestBlock  uint64 // Highest alleged block number in the chain
 	PulledStates  uint64 // Number of state trie entries already downloaded
 	KnownStates   uint64 // Total number of state trie entries known about
+
+	// "transaction indexing" fields
+	TxIndexFinishedBlocks  uint64 // Number of blocks whose transactions are already indexed
+	TxIndexRemainingBlocks uint64 // Number of blocks whose transactions are not indexed yet
 }
 
 // ChainSyncReader wraps access to the node's current sync status. If there's no

internal/ctxcapi/api.go

@@ -50,6 +50,8 @@ const (
 	defaultGasPrice = params.GWei
 )
 
+var errTxNotFound = errors.New("transaction not found")
+
 // PublicCortexAPI provides an API to access Cortex related information.
 // It offers only methods that operate on public data that is freely available to anyone.
 type PublicCortexAPI struct {
@@ -79,8 +81,8 @@ func (s *PublicCortexAPI) ProtocolVersion() hexutil.Uint {
 // - highestBlock:  block number of the highest block header this node has received from peers
 // - pulledStates:  number of state entries processed until now
 // - knownStates:   number of known state entries that still need to be pulled
-func (s *PublicCortexAPI) Syncing() (any, error) {
-	progress := s.b.SyncProgress()
+func (s *PublicCortexAPI) Syncing(ctx context.Context) (any, error) {
+	progress := s.b.SyncProgress(ctx)
 
 	// Return not syncing if the synchronisation already completed
 	if progress.CurrentBlock >= progress.HighestBlock {
@@ -1328,9 +1330,18 @@ func (s *PublicTransactionPoolAPI) GetTransactionCount(ctx context.Context, addr
 // GetTransactionByHash returns the transaction for the given hash
 func (s *PublicTransactionPoolAPI) GetTransactionByHash(ctx context.Context, hash common.Hash) (*RPCTransaction, error) {
 	// Try to return an already finalized transaction
-	tx, blockHash, blockNumber, index, err := s.b.GetTransaction(ctx, hash)
-	if err != nil {
-		return nil, err
+	found, tx, blockHash, blockNumber, index := s.b.GetTransaction(hash)
+	if !found {
+		// No finalized transaction, try to retrieve it from the pool
+		if tx := s.b.GetPoolTransaction(hash); tx != nil {
+			return newRPCPendingTransaction(tx), nil
+		}
+		// If also not in the pool there is a chance the tx indexer is still in progress.
+		if !s.b.TxIndexDone() {
+			return nil, errTxNotFound
+		}
+		// If the transaction is not found in the pool and the indexer is done, return nil
+		return nil, nil
 	}
 	if tx != nil {
 		return newRPCTransaction(tx, blockHash, blockNumber, index), nil
@@ -1347,9 +1358,15 @@ func (s *PublicTransactionPoolAPI) GetTransactionByHash(ctx context.Context, has
 // GetRawTransactionByHash returns the bytes of the transaction for the given hash.
 func (s *PublicTransactionPoolAPI) GetRawTransactionByHash(ctx context.Context, hash common.Hash) (hexutil.Bytes, error) {
 	// Retrieve a finalized transaction, or a pooled otherwise
-	tx, _, _, _, err := s.b.GetTransaction(ctx, hash)
-	if err != nil {
-		return nil, err
+	found, tx, _, _, _ := s.b.GetTransaction(hash)
+	if !found {
+		if tx = s.b.GetPoolTransaction(hash); tx != nil {
+			return tx.MarshalBinary()
+		}
+		if !s.b.TxIndexDone() {
+			return nil, errTxNotFound
+		}
+		return nil, nil
 	}
 	if tx == nil {
 		if tx = s.b.GetPoolTransaction(hash); tx == nil {