From a4bb72ec3e818f31e4bf5eda9c5d538c2008267d Mon Sep 17 00:00:00 2001 From: Gary Rong Date: Fri, 19 Jan 2024 15:10:01 +0800 Subject: [PATCH] core, eth, graphql, interfaces: improve eth.syncing --- core/blockchain.go | 58 +++++++++++++++++++++++---------------- core/blockchain_reader.go | 10 +++++-- core/blockchain_test.go | 10 +++++++ eth/api_backend.go | 7 ++++- eth/backend.go | 2 +- eth/downloader/api.go | 45 ++++++++++++++++++++++++++++-- eth/tracers/api_test.go | 2 +- ethstats/ethstats.go | 4 +-- graphql/graphql.go | 10 ++++++- interfaces.go | 12 ++++++++ internal/ethapi/api.go | 34 ++++++++++++----------- 11 files changed, 143 insertions(+), 51 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index fb7adf2444..e413624932 100644 --- a/core/blockchain.go +++ b/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 - tail = rawdb.ReadTxIndexTail(bc.db) + 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: diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 7e4eb8daa0..fbec721e49 100644 --- a/core/blockchain_reader.go +++ b/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)) +} diff --git a/core/blockchain_test.go b/core/blockchain_test.go index c1286705b6..71260e44a0 100644 --- a/core/blockchain_test.go +++ b/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() diff --git a/eth/api_backend.go b/eth/api_backend.go index c2019c3dd3..0edcce5c87 100644 --- a/eth/api_backend.go +++ b/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) { diff --git a/eth/backend.go b/eth/backend.go index 774ffaf248..aff23a910b 100644 --- a/eth/backend.go +++ b/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), diff --git a/eth/downloader/api.go b/eth/downloader/api.go index 606c6d4e7e..6d6a3f943a 100644 --- a/eth/downloader/api.go +++ b/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 + } } } } diff --git a/eth/tracers/api_test.go b/eth/tracers/api_test.go index c72e5eaaac..8aaa20fce5 100644 --- a/eth/tracers/api_test.go +++ b/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 { diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go index 75d0faac54..29559991be 100644 --- a/ethstats/ethstats.go +++ b/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") diff --git a/graphql/graphql.go b/graphql/graphql.go index e76ddac6bc..bf65b6544c 100644 --- a/graphql/graphql.go +++ b/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 diff --git a/interfaces.go b/interfaces.go index 1892309ed3..c6aee295ee 100644 --- a/interfaces.go +++ b/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 diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index f2ba2b4aac..78522c4f73 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -133,26 +133,28 @@ 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 return map[string]interface{}{ - "startingBlock": hexutil.Uint64(progress.StartingBlock), - "currentBlock": hexutil.Uint64(progress.CurrentBlock), - "highestBlock": hexutil.Uint64(progress.HighestBlock), - "syncedAccounts": hexutil.Uint64(progress.SyncedAccounts), - "syncedAccountBytes": hexutil.Uint64(progress.SyncedAccountBytes), - "syncedBytecodes": hexutil.Uint64(progress.SyncedBytecodes), - "syncedBytecodeBytes": hexutil.Uint64(progress.SyncedBytecodeBytes), - "syncedStorage": hexutil.Uint64(progress.SyncedStorage), - "syncedStorageBytes": hexutil.Uint64(progress.SyncedStorageBytes), - "healedTrienodes": hexutil.Uint64(progress.HealedTrienodes), - "healedTrienodeBytes": hexutil.Uint64(progress.HealedTrienodeBytes), - "healedBytecodes": hexutil.Uint64(progress.HealedBytecodes), - "healedBytecodeBytes": hexutil.Uint64(progress.HealedBytecodeBytes), - "healingTrienodes": hexutil.Uint64(progress.HealingTrienodes), - "healingBytecode": hexutil.Uint64(progress.HealingBytecode), + "startingBlock": hexutil.Uint64(progress.StartingBlock), + "currentBlock": hexutil.Uint64(progress.CurrentBlock), + "highestBlock": hexutil.Uint64(progress.HighestBlock), + "syncedAccounts": hexutil.Uint64(progress.SyncedAccounts), + "syncedAccountBytes": hexutil.Uint64(progress.SyncedAccountBytes), + "syncedBytecodes": hexutil.Uint64(progress.SyncedBytecodes), + "syncedBytecodeBytes": hexutil.Uint64(progress.SyncedBytecodeBytes), + "syncedStorage": hexutil.Uint64(progress.SyncedStorage), + "syncedStorageBytes": hexutil.Uint64(progress.SyncedStorageBytes), + "healedTrienodes": hexutil.Uint64(progress.HealedTrienodes), + "healedTrienodeBytes": hexutil.Uint64(progress.HealedTrienodeBytes), + "healedBytecodes": hexutil.Uint64(progress.HealedBytecodes), + "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 }