core, eth, graphql, interfaces: improve eth.syncing

core/blockchain.go

@@ -192,19 +192,15 @@ type txLookup struct {
 	transaction *types.Transaction
 }
 
-// txIndexProgress is the struct describing the progress for transaction indexing.
-type txIndexProgress struct {
-	head    uint64 // the current chain head
-	indexed uint64 // the number of blocks have been indexed
-	limit   uint64 // the number of blocks required for transaction indexing(0 means the whole chain)
+// TxIndexProgress is the struct describing the progress for transaction indexing.
+type TxIndexProgress struct {
+	Indexed   uint64 // number of blocks whose transactions are indexed
+	Remaining uint64 // number of blocks whose transactions are not indexed yet
 }
 
-// done returns an indicator if the transaction indexing is finished.
-func (prog txIndexProgress) done() bool {
-	if prog.limit == 0 {
-		return prog.indexed == (prog.head + 1) // genesis included
-	}
-	return prog.indexed >= prog.limit
+// Done returns an indicator if the transaction indexing is finished.
+func (prog TxIndexProgress) Done() bool {
+	return prog.Remaining == 0
 }
 
 // BlockChain represents the canonical chain given a database with a genesis
@@ -248,6 +244,7 @@ type BlockChain struct {
 	chainHeadFeed event.Feed
 	logsFeed      event.Feed
 	blockProcFeed event.Feed
+	txIndexFeed   event.Feed
 	scope         event.SubscriptionScope
 	genesisBlock  *types.Block
 
@@ -273,7 +270,7 @@ type BlockChain struct {
 	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
+	txIndexProgCh chan chan TxIndexProgress // chan for querying the progress of transaction indexing
 
 	engine     consensus.Engine
 	validator  Validator // Block and state validator interface
@@ -322,7 +319,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 		blockCache:    lru.NewCache[common.Hash, *types.Block](blockCacheLimit),
 		txLookupCache: lru.NewCache[common.Hash, txLookup](txLookupCacheLimit),
 		futureBlocks:  lru.NewCache[common.Hash, *types.Block](maxFutureBlocks),
-		txIndexProgCh: make(chan chan txIndexProgress),
+		txIndexProgCh: make(chan chan TxIndexProgress),
 		engine:        engine,
 		vmConfig:      vmConfig,
 	}
@@ -2451,29 +2448,38 @@ func (bc *BlockChain) indexBlocks(tail *uint64, head uint64, done chan struct{})
 }
 
 // reportTxIndexProgress returns the tx indexing progress.
-func (bc *BlockChain) reportTxIndexProgress(head uint64) txIndexProgress {
+func (bc *BlockChain) reportTxIndexProgress(head uint64) TxIndexProgress {
 	var (
-		indexed uint64
+		remaining uint64
 		tail      = rawdb.ReadTxIndexTail(bc.db)
 	)
+	total := bc.txLookupLimit
+	if bc.txLookupLimit == 0 {
+		total = head + 1 // genesis included
+	}
+	var indexed uint64
 	if tail != nil {
 		indexed = head - *tail + 1
 	}
-	return txIndexProgress{
-		head:    head,
-		indexed: indexed,
-		limit:   bc.txLookupLimit,
+	// The value of indexed might be larger than total if some blocks need
+	// to unindexed, avoiding a negative remaining.
+	if indexed < total {
+		remaining = total - indexed
+	}
+	return TxIndexProgress{
+		Indexed:   indexed,
+		Remaining: remaining,
 	}
 }
 
-// askTxIndexProgress retrieves the tx indexing progress.
-func (bc *BlockChain) askTxIndexProgress() (txIndexProgress, error) {
-	ch := make(chan txIndexProgress, 1)
+// TxIndexProgress retrieves the tx indexing progress.
+func (bc *BlockChain) TxIndexProgress() (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")
+		return TxIndexProgress{}, errors.New("blockchain is closed")
 	}
 }
 
@@ -2521,6 +2527,10 @@ func (bc *BlockChain) maintainTxIndex() {
 			lastHead = head.Block.NumberU64()
 		case <-done:
 			done = nil
+
+			if bc.reportTxIndexProgress(lastHead).Done() {
+				bc.txIndexFeed.Send(true)
+			}
 		case ch := <-bc.txIndexProgCh:
 			ch <- bc.reportTxIndexProgress(lastHead)
 		case <-bc.quit:

core/blockchain_reader.go

@@ -272,13 +272,13 @@ func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLoo
 	}
 	tx, blockHash, blockNumber, txIndex := rawdb.ReadTransaction(bc.db, hash)
 	if tx == nil {
-		progress, err := bc.askTxIndexProgress()
+		progress, err := bc.TxIndexProgress()
 		if err != nil {
 			return nil, nil, nil
 		}
 		// The transaction indexing is not finished yet, returning an
 		// error to explicitly indicate it.
-		if !progress.done() {
+		if !progress.Done() {
 			return nil, nil, errors.New("transaction indexing still in progress")
 		}
 		// The transaction is already indexed, the transaction is either
@@ -444,3 +444,9 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
 func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
 	return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
 }
+
+// SubscribeTxIndexEvent registers a subscription of bool where true means
+// transaction indexing has finished.
+func (bc *BlockChain) SubscribeTxIndexEvent(ch chan<- bool) event.Subscription {
+	return bc.scope.Track(bc.txIndexFeed.Subscribe(ch))
+}

core/blockchain_test.go

@@ -4075,6 +4075,12 @@ func TestTxIndexer(t *testing.T) {
 		}
 		verifyRange(db, *tail, 128, true)
 	}
+	verifyProgress := func(chain *BlockChain) {
+		prog := chain.reportTxIndexProgress(128)
+		if !prog.Done() {
+			t.Fatalf("Expect fully indexed")
+		}
+	}
 
 	var cases = []struct {
 		limitA uint64
@@ -4204,19 +4210,23 @@ func TestTxIndexer(t *testing.T) {
 		chain, _ := NewBlockChain(db, nil, gspec, nil, engine, vm.Config{}, nil, &c.limitA)
 		chain.indexBlocks(nil, 128, make(chan struct{}))
 		verify(db, c.tailA)
+		verifyProgress(chain)
 
 		chain.SetTxLookupLimit(c.limitB)
 		chain.indexBlocks(rawdb.ReadTxIndexTail(db), 128, make(chan struct{}))
 		verify(db, c.tailB)
+		verifyProgress(chain)
 
 		chain.SetTxLookupLimit(c.limitC)
 		chain.indexBlocks(rawdb.ReadTxIndexTail(db), 128, make(chan struct{}))
 		verify(db, c.tailC)
+		verifyProgress(chain)
 
 		// Recover all indexes
 		chain.SetTxLookupLimit(0)
 		chain.indexBlocks(rawdb.ReadTxIndexTail(db), 128, make(chan struct{}))
 		verify(db, 0)
+		verifyProgress(chain)
 
 		chain.Stop()
 		db.Close()

eth/api_backend.go

@@ -354,7 +354,12 @@ func (b *EthAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.S
 }
 
 func (b *EthAPIBackend) SyncProgress() ethereum.SyncProgress {
-	return b.eth.Downloader().Progress()
+	prog := b.eth.Downloader().Progress()
+	if txProg, err := b.eth.blockchain.TxIndexProgress(); err == nil {
+		prog.TxIndexFinishedBlocks = txProg.Indexed
+		prog.TxIndexRemainingBlocks = txProg.Remaining
+	}
+	return prog
 }
 
 func (b *EthAPIBackend) SuggestGasTipCap(ctx context.Context) (*big.Int, error) {

eth/backend.go

@@ -322,7 +322,7 @@ func (s *Ethereum) APIs() []rpc.API {
 			Service:   NewMinerAPI(s),
 		}, {
 			Namespace: "eth",
-			Service:   downloader.NewDownloaderAPI(s.handler.downloader, s.eventMux),
+			Service:   downloader.NewDownloaderAPI(s.handler.downloader, s.blockchain, s.eventMux),
 		}, {
 			Namespace: "admin",
 			Service:   NewAdminAPI(s),

eth/downloader/api.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/rpc"
 )
@@ -29,6 +30,7 @@ import (
 // It offers only methods that operates on data that can be available to anyone without security risks.
 type DownloaderAPI struct {
 	d                         *Downloader
+	chain                     *core.BlockChain
 	mux                       *event.TypeMux
 	installSyncSubscription   chan chan interface{}
 	uninstallSyncSubscription chan *uninstallSyncSubscriptionRequest
@@ -38,9 +40,10 @@ type DownloaderAPI struct {
 // listens for events from the downloader through the global event mux. In case it receives one of
 // these events it broadcasts it to all syncing subscriptions that are installed through the
 // installSyncSubscription channel.
-func NewDownloaderAPI(d *Downloader, m *event.TypeMux) *DownloaderAPI {
+func NewDownloaderAPI(d *Downloader, chain *core.BlockChain, m *event.TypeMux) *DownloaderAPI {
 	api := &DownloaderAPI{
 		d:                         d,
+		chain:                     chain,
 		mux:                       m,
 		installSyncSubscription:   make(chan chan interface{}),
 		uninstallSyncSubscription: make(chan *uninstallSyncSubscriptionRequest),
@@ -57,7 +60,10 @@ func (api *DownloaderAPI) eventLoop() {
 	var (
 		sub               = api.mux.Subscribe(StartEvent{}, DoneEvent{}, FailedEvent{})
 		syncSubscriptions = make(map[chan interface{}]struct{})
+		txIndexCh         = make(chan bool, 1)
 	)
+	txIndexSub := api.chain.SubscribeTxIndexEvent(txIndexCh)
+	defer txIndexSub.Unsubscribe()
 
 	for {
 		select {
@@ -70,21 +76,54 @@ func (api *DownloaderAPI) eventLoop() {
 			if event == nil {
 				return
 			}
-
 			var notification interface{}
+
 			switch event.Data.(type) {
 			case StartEvent:
+				prog := api.d.Progress()
+				txProg, err := api.chain.TxIndexProgress()
+				if err == nil {
+					prog.TxIndexFinishedBlocks = txProg.Indexed
+					prog.TxIndexRemainingBlocks = txProg.Remaining
+				}
 				notification = &SyncingResult{
 					Syncing: true,
-					Status:  api.d.Progress(),
+					Status:  prog,
 				}
 			case DoneEvent, FailedEvent:
 				notification = false
+
+				txProg, err := api.chain.TxIndexProgress()
+				if err == nil && !txProg.Done() {
+					prog := api.d.Progress()
+					prog.TxIndexFinishedBlocks = txProg.Indexed
+					prog.TxIndexRemainingBlocks = txProg.Remaining
+					notification = &SyncingResult{
+						Syncing: true,
+						Status:  prog,
+					}
+				}
 			}
 			// broadcast
 			for c := range syncSubscriptions {
 				c <- notification
 			}
+		case status := <-txIndexCh:
+			if !status {
+				continue
+			}
+			prog := api.d.Progress()
+			txProg, err := api.chain.TxIndexProgress()
+			if err == nil && !txProg.Done() {
+				prog.TxIndexFinishedBlocks = txProg.Indexed
+				prog.TxIndexRemainingBlocks = txProg.Remaining
+			}
+			if !prog.Done() {
+				continue
+			}
+			for c := range syncSubscriptions {
+				c <- true
+			}
 		}
 	}
 }

eth/tracers/api_test.go

@@ -115,7 +115,7 @@ func (b *testBackend) BlockByNumber(ctx context.Context, number rpc.BlockNumber)
 
 func (b *testBackend) GetTransaction(ctx context.Context, txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64, error) {
 	tx, hash, blockNumber, index := rawdb.ReadTransaction(b.chaindb, txHash)
-	return true, tx, hash, blockNumber, index, nil
+	return tx != nil, tx, hash, blockNumber, index, nil
 }
 
 func (b *testBackend) RPCGasCap() uint64 {

ethstats/ethstats.go

@@ -792,7 +792,7 @@ func (s *Service) reportStats(conn *connWrapper) error {
 		}
 
 		sync := fullBackend.SyncProgress()
-		syncing = fullBackend.CurrentHeader().Number.Uint64() >= sync.HighestBlock
+		syncing = !sync.Done()
 
 		price, _ := fullBackend.SuggestGasTipCap(context.Background())
 		gasprice = int(price.Uint64())
@@ -801,7 +801,7 @@ func (s *Service) reportStats(conn *connWrapper) error {
 		}
 	} else {
 		sync := s.backend.SyncProgress()
-		syncing = s.backend.CurrentHeader().Number.Uint64() >= sync.HighestBlock
+		syncing = !sync.Done()
 	}
 	// Assemble the node stats and send it to the server
 	log.Trace("Sending node details to ethstats")

graphql/graphql.go

@@ -1509,6 +1509,12 @@ func (s *SyncState) HealingTrienodes() hexutil.Uint64 {
 func (s *SyncState) HealingBytecode() hexutil.Uint64 {
 	return hexutil.Uint64(s.progress.HealingBytecode)
 }
+func (s *SyncState) TxIndexFinishedBlocks() hexutil.Uint64 {
+	return hexutil.Uint64(s.progress.TxIndexFinishedBlocks)
+}
+func (s *SyncState) TxIndexRemainingBlocks() hexutil.Uint64 {
+	return hexutil.Uint64(s.progress.TxIndexRemainingBlocks)
+}
 
 // Syncing returns false in case the node is currently not syncing with the network. It can be up-to-date or has not
 // yet received the latest block headers from its pears. In case it is synchronizing:
@@ -1527,11 +1533,13 @@ func (s *SyncState) HealingBytecode() hexutil.Uint64 {
 // - healedBytecodeBytes: number of bytecodes persisted to disk
 // - healingTrienodes:    number of state trie nodes pending
 // - healingBytecode:     number of bytecodes pending
+// - txIndexFinishedBlocks:  number of blocks whose transactions are indexed
+// - txIndexRemainingBlocks: number of blocks whose transactions are not indexed yet
 func (r *Resolver) Syncing() (*SyncState, error) {
 	progress := r.backend.SyncProgress()
 
 	// Return not syncing if the synchronisation already completed
-	if progress.CurrentBlock >= progress.HighestBlock {
+	if progress.Done() {
 		return nil, nil
 	}
 	// Otherwise gather the block sync stats

interfaces.go

@@ -120,6 +120,18 @@ type SyncProgress struct {
 
 	HealingTrienodes uint64 // Number of state trie nodes pending
 	HealingBytecode  uint64 // Number of bytecodes pending
+
+	// "transaction indexing" fields
+	TxIndexFinishedBlocks  uint64 // Number of blocks whose transactions are already indexed
+	TxIndexRemainingBlocks uint64 // Number of blocks whose transactions are not indexed yet
+}
+
+// Done returns the indicator if the initial sync is finished or not.
+func (prog SyncProgress) Done() bool {
+	if prog.CurrentBlock < prog.HighestBlock {
+		return false
+	}
+	return prog.TxIndexRemainingBlocks == 0
 }
 
 // ChainSyncReader wraps access to the node's current sync status. If there's no

internal/ethapi/api.go

@@ -133,7 +133,7 @@ func (s *EthereumAPI) Syncing() (interface{}, error) {
 	progress := s.b.SyncProgress()
 
 	// Return not syncing if the synchronisation already completed
-	if progress.CurrentBlock >= progress.HighestBlock {
+	if progress.Done() {
 		return false, nil
 	}
 	// Otherwise gather the block sync stats
@@ -153,6 +153,8 @@ func (s *EthereumAPI) Syncing() (interface{}, error) {
 		"healedBytecodeBytes":    hexutil.Uint64(progress.HealedBytecodeBytes),
 		"healingTrienodes":       hexutil.Uint64(progress.HealingTrienodes),
 		"healingBytecode":        hexutil.Uint64(progress.HealingBytecode),
+		"txIndexFinishedBlocks":  hexutil.Uint64(progress.TxIndexFinishedBlocks),
+		"txIndexRemainingBlocks": hexutil.Uint64(progress.TxIndexRemainingBlocks),
 	}, nil
 }