From f7792c34e9366bd5978d56e10e54b3bb20581096 Mon Sep 17 00:00:00 2001 From: zsfelfoldi Date: Mon, 29 Sep 2025 06:04:23 +0200 Subject: [PATCH] core: add Indexer interface and index server logic --- core/blockchain.go | 32 ++- core/blockchain_reader.go | 8 + core/index_server.go | 592 ++++++++++++++++++++++++++++++++++++++ core/index_server_test.go | 214 ++++++++++++++ 4 files changed, 845 insertions(+), 1 deletion(-) create mode 100644 core/index_server.go create mode 100644 core/index_server_test.go diff --git a/core/blockchain.go b/core/blockchain.go index b7acd12aca..4ed8dddce7 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -308,6 +308,7 @@ type BlockChain struct { blockProcCounter int32 scope event.SubscriptionScope genesisBlock *types.Block + indexServers indexServers // This mutex synchronizes chain write operations. // Readers don't need to take it, they can just read the database. @@ -540,9 +541,15 @@ func NewBlockChain(db ethdb.Database, genesis *Genesis, engine consensus.Engine, log.Info("Failed to setup size tracker", "err", err) } } + bc.indexServers.init(bc) return bc, nil } +// RegisterIndexer registers a new indexer to the chain. +func (bc *BlockChain) RegisterIndexer(indexer Indexer, name string) { + bc.indexServers.register(indexer, name) +} + func (bc *BlockChain) setupSnapshot() { // Short circuit if the chain is established with path scheme, as the // state snapshot has been integrated into path database natively. @@ -655,6 +662,7 @@ func (bc *BlockChain) loadLastState() error { if head := rawdb.ReadFinalizedBlockHash(bc.db); head != (common.Hash{}) { if block := bc.GetBlockByHash(head); block != nil { bc.currentFinalBlock.Store(block.Header()) + bc.indexServers.setFinalBlock(block.NumberU64()) headFinalizedBlockGauge.Update(int64(block.NumberU64())) bc.currentSafeBlock.Store(block.Header()) headSafeBlockGauge.Update(int64(block.NumberU64())) @@ -702,6 +710,7 @@ func (bc *BlockChain) initializeHistoryPruning(latest uint64) error { return errors.New("unexpected database tail") } bc.historyPrunePoint.Store(predefinedPoint) + bc.indexServers.setHistoryCutoff(predefinedPoint.BlockNumber) return nil case history.KeepPostMerge: @@ -723,6 +732,7 @@ func (bc *BlockChain) initializeHistoryPruning(latest uint64) error { return errors.New("unexpected database tail") } bc.historyPrunePoint.Store(predefinedPoint) + bc.indexServers.setHistoryCutoff(predefinedPoint.BlockNumber) return nil default: @@ -785,9 +795,11 @@ func (bc *BlockChain) SetFinalized(header *types.Header) { if header != nil { rawdb.WriteFinalizedBlockHash(bc.db, header.Hash()) headFinalizedBlockGauge.Update(int64(header.Number.Uint64())) + bc.indexServers.setFinalBlock(header.Number.Uint64()) } else { rawdb.WriteFinalizedBlockHash(bc.db, common.Hash{}) headFinalizedBlockGauge.Update(0) + bc.indexServers.setFinalBlock(0) } } @@ -1094,6 +1106,7 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha bc.receiptsCache.Purge() bc.blockCache.Purge() bc.txLookupCache.Purge() + bc.indexServers.revert(bc.CurrentBlock()) // Clear safe block, finalized block if needed if safe := bc.CurrentSafeBlock(); safe != nil && head < safe.Number.Uint64() { @@ -1150,6 +1163,7 @@ func (bc *BlockChain) Reset() error { // ResetWithGenesisBlock purges the entire blockchain, restoring it to the // specified genesis state. func (bc *BlockChain) ResetWithGenesisBlock(genesis *types.Block) error { + bc.indexServers.revert(genesis.Header()) // Dump the entire block chain and purge the caches if err := bc.SetHead(0); err != nil { return err @@ -1166,6 +1180,7 @@ func (bc *BlockChain) ResetWithGenesisBlock(genesis *types.Block) error { log.Crit("Failed to write genesis block", "err", err) } bc.writeHeadBlock(genesis) + bc.indexServers.broadcast(genesis.Header()) // Last update all in-memory chain markers bc.genesisBlock = genesis @@ -1282,6 +1297,7 @@ func (bc *BlockChain) stopWithoutSaving() { // Stop stops the blockchain service. If any imports are currently in progress // it will abort them using the procInterrupt. func (bc *BlockChain) Stop() { + bc.indexServers.stop() bc.stopWithoutSaving() // Ensure that the entirety of the state snapshot is journaled to disk. @@ -1583,6 +1599,7 @@ func (bc *BlockChain) writeKnownBlock(block *types.Block) error { } } bc.writeHeadBlock(block) + bc.indexServers.broadcast(block.Header()) return nil } @@ -1598,7 +1615,14 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types. // should be written atomically. BlockBatch is used for containing all components. blockBatch := bc.db.NewBatch() rawdb.WriteBlock(blockBatch, block) - rawdb.WriteReceipts(blockBatch, block.Hash(), block.NumberU64(), receipts) + blockHash := block.Hash() + bc.blockCache.Add(blockHash, block) + rawdb.WriteReceipts(blockBatch, blockHash, block.NumberU64(), receipts) + if receipts != nil { + bc.receiptsCache.Add(blockHash, receipts) + } else { + bc.receiptsCache.Add(blockHash, []*types.Receipt{}) + } rawdb.WritePreimages(blockBatch, statedb.Preimages()) if err := blockBatch.Write(); err != nil { log.Crit("Failed to write block into disk", "err", err) @@ -1689,6 +1713,7 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types // Set new head. bc.writeHeadBlock(block) + bc.indexServers.broadcast(block.Header()) bc.chainFeed.Send(ChainEvent{ Header: block.Header(), @@ -1760,10 +1785,12 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness } if atomic.AddInt32(&bc.blockProcCounter, 1) == 1 { + bc.indexServers.setBlockProcessing(true) bc.blockProcFeed.Send(true) } defer func() { if atomic.AddInt32(&bc.blockProcCounter, -1) == 0 { + bc.indexServers.setBlockProcessing(false) bc.blockProcFeed.Send(false) } }() @@ -2426,6 +2453,7 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Header) error return errInvalidNewChain } } + bc.indexServers.revert(commonBlock) // Ensure the user sees large reorgs if len(oldChain) > 0 && len(newChain) > 0 { logFn := log.Info @@ -2523,6 +2551,7 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Header) error } // Update the head block bc.writeHeadBlock(block) + bc.indexServers.broadcast(block.Header()) } if len(rebirthLogs) > 0 { bc.logsFeed.Send(rebirthLogs) @@ -2598,6 +2627,7 @@ func (bc *BlockChain) SetCanonical(head *types.Block) (common.Hash, error) { } } bc.writeHeadBlock(head) + bc.indexServers.broadcast(head.Header()) // Emit events receipts, logs := bc.collectReceiptsAndLogs(head, false) diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 4894523b0e..a79a89c07d 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -266,6 +266,14 @@ func (bc *BlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts { if !ok { return nil } + return bc.GetReceipts(hash, number) +} + +// GetReceipts retrieves the receipts for all transactions in a given block. +func (bc *BlockChain) GetReceipts(hash common.Hash, number uint64) types.Receipts { + if receipts, ok := bc.receiptsCache.Get(hash); ok { + return receipts + } header := bc.GetHeader(hash, number) if header == nil { return nil diff --git a/core/index_server.go b/core/index_server.go new file mode 100644 index 0000000000..73070c4d58 --- /dev/null +++ b/core/index_server.go @@ -0,0 +1,592 @@ +// Copyright 2025 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package core implements the Ethereum consensus protocol. +package core + +import ( + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/log" +) + +const ( + busyDelay = time.Second // indexer status polling frequency when not ready + maxHistoricPrefetch = 16 // size of block data pre-fetch queue + logFrequency = time.Second * 20 // log info frequency during long indexing/unindexing process + headLogDelay = time.Second // head indexing log info delay (do not log if finished faster) +) + +type Indexer interface { + // AddBlockData delivers a header and receipts belonging to a block that is + // either a direct descendant of the latest delivered head or the first one + // in the last requested range. + // The current ready/busy status and the requested historic range are returned. + // Note that the indexer should never block even if it is busy processing. + // It is allowed to re-request the delivered blocks later if the indexer could + // not process them when first delivered. + AddBlockData(header *types.Header, receipts types.Receipts) (ready bool, needBlocks common.Range[uint64]) + // Revert rewinds the index to the given head block number. Subsequent + // AddBlockData calls will deliver blocks starting from this point. + Revert(blockNumber uint64) + // Status returns the current ready/busy status and the requested historic range. + // Only the new head blocks are delivered if the indexer reports busy status. + Status() (ready bool, needBlocks common.Range[uint64]) + // SetHistoryCutoff signals the historical cutoff point to the indexer. + // Note that any block number that is consistently being requested in the + // needBlocks response that is not older than the cutoff point is guaranteed + // to be delivered eventually. If the required data belonging to certain + // block numbers is missing then the cutoff point is moved after the missing + // section in order to maintain this guarantee. + SetHistoryCutoff(blockNumber uint64) + // SetFinalized signals the latest finalized block number to the indexer. + SetFinalized(blockNumber uint64) + // Suspended signals to the indexer that block processing has started and + // any non-essential asynchronous tasks of the indexer should be suspended. + // The next AddBlockData call signals the end of the suspended state. + // Note that if multiple blocks are inserted then the indexer is only + // suspended once, before the first block processing begins, so according + // to the rule above it will not be suspended while processing the rest of + // the batch. This behavior should be fine because indexing can happen in + // parallel with forward syncing, the purpose of the suspend mechanism is + // to handle historical index backfilling with a lower priority so that it + // does not increase block latency. + Suspended() + // Stop initiates indexer shutdown. No subsequent calls are made through this + // interface after Stop. + Stop() +} + +// indexServers operates as a part of BlockChain and can serve multiple chain +// indexers that implement the Indexer interface. +type indexServers struct { + lock sync.Mutex + servers []*indexServer + chain *BlockChain + + lastHead *types.Header + lastHeadReceipts types.Receipts + finalBlock, historyCutoff uint64 + + closeCh chan struct{} + closeWg sync.WaitGroup +} + +// init initializes indexServers. +func (f *indexServers) init(chain *BlockChain) { + f.lock.Lock() + defer f.lock.Unlock() + + f.chain = chain + f.lastHead = chain.CurrentBlock() + f.closeCh = make(chan struct{}) +} + +// stop shuts down all registered Indexers and their serving goroutines. +func (f *indexServers) stop() { + f.lock.Lock() + defer f.lock.Unlock() + + close(f.closeCh) + f.closeWg.Wait() + f.servers = nil +} + +// register adds a new Indexer to the chain. +func (f *indexServers) register(indexer Indexer, name string) { + f.lock.Lock() + defer f.lock.Unlock() + + server := &indexServer{ + parent: f, + indexer: indexer, + sendTimer: time.NewTimer(0), + lastHead: f.lastHead, + name: name, + statusCh: make(chan indexerStatus, 1), + blockDataCh: make(chan blockData, maxHistoricPrefetch), + suspendCh: make(chan bool, 1), + } + f.servers = append(f.servers, server) + f.closeWg.Add(2) + indexer.SetHistoryCutoff(f.historyCutoff) + indexer.SetFinalized(f.finalBlock) + if f.lastHead != nil { + server.status.ready, server.status.needBlocks = indexer.AddBlockData(f.lastHead, f.lastHeadReceipts) + server.updateStatus() + } + go server.historicReadLoop() + go server.historicSendLoop() +} + +// broadcast sends a new head block to all registered Indexer instances. +func (f *indexServers) broadcast(header *types.Header) { + f.lock.Lock() + defer f.lock.Unlock() + + blockReceipts := f.chain.GetReceipts(header.Hash(), header.Number.Uint64()) + if blockReceipts == nil { + log.Error("Receipts belonging to new head are missing", "number", header.Number, "hash", header.Hash()) + return + } + f.lastHead, f.lastHeadReceipts = header, blockReceipts + for _, server := range f.servers { + server.sendHeadBlockData(header, blockReceipts) + } +} + +// revert notifies all registered Indexer instances about the chain being rolled +// back to the given head or last common ancestor. +func (f *indexServers) revert(header *types.Header) { + f.lock.Lock() + defer f.lock.Unlock() + + for _, server := range f.servers { + server.revert(header) + } +} + +// setFinalBlock notifies all Indexer instances about the latest finalized block. +func (f *indexServers) setFinalBlock(blockNumber uint64) { + f.lock.Lock() + defer f.lock.Unlock() + + if f.finalBlock == blockNumber { + return + } + f.finalBlock = blockNumber + for _, server := range f.servers { + server.setFinalBlock(blockNumber) + } +} + +// setHistoryCutoff notifies all Indexer instances about the history cutoff point. +// The indexers cannot expect any data being delivered if needBlocks.First() is +// before this point. +func (f *indexServers) setHistoryCutoff(blockNumber uint64) { + f.lock.Lock() + defer f.lock.Unlock() + + if f.historyCutoff == blockNumber { + return + } + f.historyCutoff = blockNumber + for _, server := range f.servers { + server.setHistoryCutoff(blockNumber) + } +} + +// setBlockProcessing suspends serving historical blocks requested by the indexers +// while a chain segment is being processed and added to the chain. +func (f *indexServers) setBlockProcessing(processing bool) { + f.lock.Lock() + defer f.lock.Unlock() + + for _, server := range f.servers { + server.setBlockProcessing(processing) + } +} + +// indexServer sends updates to a single Indexer instance. It sends all new heads +// and reorg events, and also sends historical block data upon request. +// It guarantees that Indexer functions are never called concurrently and also +// they always present a consistent view of the chain to the indexer. +type indexServer struct { + lock sync.Mutex + parent *indexServers + indexer Indexer // always call under mutex lock; never call after stopped + stopped bool + + lastHead *types.Header + status indexerStatus + statusCh chan indexerStatus + blockDataCh chan blockData + suspendCh chan bool + testSuspendHookCh chan struct{} // initialized by test, capacity = 1 + lastRevertBlock uint64 + sendTimer *time.Timer + historyCutoff, missingBlockCutoff uint64 + + name string + processed uint64 + logged bool + startedAt, lastLoggedAt time.Time + lastHistoryErrorLog time.Time +} + +// indexerStatus represents the state of the indexer and also has fields that +// serve the coordination between historic reader and sender goroutines. +type indexerStatus struct { + ready, suspended, resetQueue bool + revertCount uint64 + needBlocks common.Range[uint64] +} + +// blockData represents the indexable data of a single block being sent from the +// reader to the sender goroutine and optionally queued in blockDataCh inbetween. +// It also returns the latest revertCount known before reading the block data, +// which allows the sender to guarantee that all sent block data is always +// consistent with the indexer's canonical chain view while the reading of block +// data can still happen asynchronously. +type blockData struct { + blockNumber, revertCount uint64 + header *types.Header + receipts types.Receipts +} + +// sendHeadBlockData immediately sends the latest head block data to the indexer +// and updates the status of the historical block data serving mechanism +// accordingly. +func (s *indexServer) sendHeadBlockData(header *types.Header, receipts types.Receipts) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.stopped { + return + } + if header.Hash() == s.lastHead.Hash() { + return + } + s.lastHead = header + s.status.ready, s.status.needBlocks = s.indexer.AddBlockData(header, receipts) + s.updateStatus() +} + +// revert immediately reverts the indexer to the given block and updates the +// status of the historical block data serving mechanism accordingly. +func (s *indexServer) revert(header *types.Header) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.stopped || s.lastHead == nil { + return + } + if header.Hash() == s.lastHead.Hash() { + return + } + blockNumber := header.Number.Uint64() + if blockNumber >= s.lastHead.Number.Uint64() { + panic("invalid indexer revert") + } + s.lastHead = header + s.status.revertCount++ + s.lastRevertBlock = blockNumber + s.updateStatus() + s.indexer.Revert(blockNumber) +} + +// historicSendLoop is the main event loop that interacts with the indexer in +// case when historical block data is requested. It sends status updates to +// the reader goroutine through statusCh and feeds the fetched data coming from +// blockDataCh into the indexer. +func (s *indexServer) historicSendLoop() { + defer func() { + s.lock.Lock() + s.indexer.Stop() + s.stopped = true + s.lock.Unlock() + s.parent.closeWg.Done() + }() + + // suspendOrStop blocks the send loop until it is unsuspended or the parent + // chain is stopped. It also notifies the indexer by calling Suspend and + // suspends the read loop through updateStatus. + suspendOrStop := func(suspended bool) bool { + if !suspended { + panic("unexpected 'false' signal on suspendCh") + } + s.lock.Lock() + s.status.suspended = true + s.updateStatus() + s.indexer.Suspended() + s.lock.Unlock() + select { + case <-s.parent.closeCh: + return true + case suspended = <-s.suspendCh: + } + if suspended { + panic("unexpected 'true' signal on suspendCh") + } + s.lock.Lock() + s.status.suspended = false + s.updateStatus() + s.lock.Unlock() + return false + } + + for { + select { + // do a separate non-blocking select to ensure that a suspend attempt + // during the previous historical AddBlockData will be catched in the + // next round. + case ch := <-s.suspendCh: + if suspendOrStop(ch) { + return + } + default: + } + select { + case <-s.parent.closeCh: + return + case ch := <-s.suspendCh: + if suspendOrStop(ch) { + return + } + case nextBlockData := <-s.blockDataCh: + s.lock.Lock() + s.status.resetQueue = true + if !s.status.needBlocks.IsEmpty() && s.status.needBlocks.First() == nextBlockData.blockNumber && + (nextBlockData.revertCount == s.status.revertCount || (nextBlockData.revertCount+1 == s.status.revertCount && nextBlockData.blockNumber <= s.lastRevertBlock)) { + s.status.ready, s.status.needBlocks = s.indexer.AddBlockData(nextBlockData.header, nextBlockData.receipts) + if nextBlockData.header != nil { + s.status.resetQueue = false + if s.status.needBlocks.IsEmpty() { + s.logDelivered(nextBlockData.blockNumber) + s.logFinished() + } else if s.status.needBlocks.First() == nextBlockData.blockNumber+1 { + s.logDelivered(nextBlockData.blockNumber) + } + } else { + s.missingBlockCutoff = max(s.missingBlockCutoff, nextBlockData.blockNumber+1) + s.indexer.SetHistoryCutoff(max(s.historyCutoff, s.missingBlockCutoff)) + s.status.ready, s.status.needBlocks = s.indexer.Status() + } + } + s.updateStatus() + s.lock.Unlock() + case <-s.sendTimer.C: + s.lock.Lock() + if !s.status.ready { + s.status.ready, s.status.needBlocks = s.indexer.Status() + s.updateStatus() + } + s.lock.Unlock() + } + } +} + +// updateStatus updates the asynchronous reader goroutine's status based on the +// latest indexer status. If necessary then it trims the needBlocks range based +// on the locally available block range. If there is already an unread status +// update waiting on statusCh then it is replaced by the new one. +func (s *indexServer) updateStatus() { + if s.status.ready || s.status.suspended { + s.sendTimer.Stop() + } else { + s.sendTimer.Reset(busyDelay) + } + var headNumber uint64 + if s.lastHead != nil { + headNumber = s.lastHead.Number.Uint64() + } + if headNumber+1 < s.status.needBlocks.AfterLast() { + s.status.needBlocks.SetLast(headNumber) + } + if s.status.needBlocks.IsEmpty() || max(s.historyCutoff, s.missingBlockCutoff) > s.status.needBlocks.First() { + s.status.needBlocks = common.Range[uint64]{} + } + select { + case <-s.statusCh: + default: + } + s.statusCh <- s.status +} + +// setBlockProcessing suspends serving historical blocks requested by the indexer +// while a chain segment is being processed and added to the chain. +func (s *indexServer) setBlockProcessing(suspended bool) { + select { + case old := <-s.suspendCh: + if old == suspended { + panic("unexpected value pulled back from suspendCh") + } + default: + // only send new suspended flag if previous (opposite) value has been + // read by the send loop already + s.suspendCh <- suspended + } + if suspended && s.testSuspendHookCh != nil { + select { + case s.testSuspendHookCh <- struct{}{}: + default: + } + } +} + +// historicReadLoop reads requested historical block data asynchronously. +// It receives indexer status updates on statusCh and sends block data to +// blockDataCh. If the latest status indicates that there the server is not +// suspended then it is guaranteed that eventually a corresponding block data +// response will be sent unless a new status update is received before this +// happens. +// Note that blockDataCh can queue multiple block data pre-fetched by +// historicReadLoop. If the requested range is changed while there is still +// queued data in the channel that corresponds to the previous requested range +// then the receiver sends a new status update with the resetQueue flag set to +// true. In this case historicReadLoop removes all remaining entries from the +// queue and starts sending block data from the beginning of the new range. +func (s *indexServer) historicReadLoop() { + defer s.parent.closeWg.Done() + + var ( + status indexerStatus + sendRange common.Range[uint64] + ) + + statusUpdated := func() { + if status.resetQueue { + // If the receiver found an item in the queue that is no longer + // relevant then we remove all remaining items first. + loop: + for { + select { + case <-s.blockDataCh: + default: + break loop + } + } + } + if !status.resetQueue && !sendRange.IsEmpty() && status.needBlocks.Includes(sendRange.First()) { + // Here we assume that the block data between needBlocks.First() and + // sendRange.First()-1 is already in the queue. + r := status.needBlocks + r.SetFirst(sendRange.First()) + sendRange = r + } else { + sendRange = status.needBlocks + } + if sendRange.Count() > maxHistoricPrefetch { + // Note: in a normal use case where needBlocks.First() is advanced + // after reading the previous item from blockDataCh, this check will + // prevent reading more data than what fits into the channel capacity. + sendRange.SetAfterLast(sendRange.First() + maxHistoricPrefetch) + } + } + + for { + if !sendRange.IsEmpty() && !status.suspended { + // Send next item to the queue. + bd := blockData{blockNumber: sendRange.First(), revertCount: status.revertCount} + if header := s.parent.chain.GetHeaderByNumber(bd.blockNumber); header != nil { + if receipts := s.parent.chain.GetReceipts(header.Hash(), bd.blockNumber); receipts != nil { + bd.header, bd.receipts = header, receipts + } else { + log.Error("Historical receipts are missing", "number", bd.blockNumber, "hash", header.Hash()) + } + } else { + log.Error("Historical header missing", "number", bd.blockNumber) + } + // Note that a response with empty block data is still sent in case of + // a read error, signaling to the sender logic that something is missing. + select { + case s.blockDataCh <- bd: + sendRange.SetFirst(bd.blockNumber + 1) + default: + // Note: in extreme corner cases where sendRange.Count() check + // does not prevent trying to overfill the channel, we simply + // reset sendRange to empty preventing more wasted reads. + // Sending the queued data will generate more status updates + // which will reinitialize sendRange once the queue is again + // below full capacity. + sendRange = common.Range[uint64]{} + } + // Keep checking status updates without blocking as long as there is + // something to do. + select { + case <-s.parent.closeCh: + return + case status = <-s.statusCh: + statusUpdated() + default: + } + } else { + // There was nothing to do; wait for a next status update. + select { + case <-s.parent.closeCh: + return + case status = <-s.statusCh: + statusUpdated() + } + } + } +} + +// logDelivered periodically prints log messages that report the current state +// of the indexing process. If should be called after processing each new block. +func (s *indexServer) logDelivered(position uint64) { + if s.processed == 0 { + s.startedAt = time.Now() + } + s.processed++ + if s.logged { + if time.Since(s.lastLoggedAt) < logFrequency { + return + } + } else { + if time.Since(s.startedAt) < headLogDelay { + return + } + s.logged = true + } + s.lastLoggedAt = time.Now() + log.Info("Generating "+s.name, "block", position, "processed", s.processed, "elapsed", time.Since(s.startedAt)) +} + +// logFinished prints a log message that report the end of the indexing process. +// Note that any log message is only printed if the process took longer than +// headLogDelay. +func (s *indexServer) logFinished() { + if s.logged { + log.Info("Finished "+s.name, "processed", s.processed, "elapsed", time.Since(s.startedAt)) + s.logged = false + } + s.processed = 0 +} + +// setFinalBlock notifies the Indexer instance about the latest finalized block. +func (s *indexServer) setFinalBlock(blockNumber uint64) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.stopped { + return + } + s.indexer.SetFinalized(blockNumber) +} + +// setHistoryCutoff notifies the Indexer instance about the history cutoff point. +// The indexer cannot expect any data being delivered if needBlocks.First() is +// before this point. +// Note that if some historical block data could not be loaded from the database +// then the historical cutoff point reported to the indexer might be modified by +// missingBlockCutoff. This workaround ensures that the indexing process does not +// get stuck permanently in case of missing data. +func (s *indexServer) setHistoryCutoff(blockNumber uint64) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.stopped { + return + } + s.historyCutoff = blockNumber + s.indexer.SetHistoryCutoff(max(s.historyCutoff, s.missingBlockCutoff)) + s.status.ready, s.status.needBlocks = s.indexer.Status() + s.updateStatus() +} diff --git a/core/index_server_test.go b/core/index_server_test.go new file mode 100644 index 0000000000..1f08ab030e --- /dev/null +++ b/core/index_server_test.go @@ -0,0 +1,214 @@ +// Copyright 2025 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package core implements the Ethereum consensus protocol. +package core + +import ( + "math/big" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus/ethash" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/params" +) + +func TestIndexServer(t *testing.T) { + ti := &testIndexer{ + t: t, + eventCh: make(chan testIndexerEvent), + statusCh: make(chan testIndexerStatus), + } + var blockchain *BlockChain + doneCh := make(chan struct{}) + expDone := func() { + select { + case <-doneCh: + case <-time.After(time.Second * 5): + t.Fatalf("Expected chain operation done but not finished yet") + } + } + testSuspendHookCh := make(chan struct{}, 1) + run := func(fn func()) { + go func() { + fn() + doneCh <- struct{}{} + }() + } + insert := func(chain []*types.Block) { + run(func() { + if i, err := blockchain.InsertChain(chain); err != nil { + t.Fatalf("failed to insert chain[%d]: %v", i, err) + } + }) + } + waitSuspend := func() { + select { + case <-testSuspendHookCh: + case <-time.After(time.Second * 5): + t.Fatalf("Expected index server suspend but suspended state not reached") + } + } + + gspec := &Genesis{ + Config: params.TestChainConfig, + BaseFee: big.NewInt(params.InitialBaseFee), + } + db := rawdb.NewMemoryDatabase() + blockchain, _ = NewBlockChain(db, gspec, ethash.NewFaker(), DefaultConfig()) + chain := []*types.Block{gspec.ToBlock()} + blocks, _ := GenerateChain(gspec.Config, chain[0], ethash.NewFaker(), db, 110, func(i int, gen *BlockGen) {}) + chain = append(chain, blocks...) + + run(func() { + blockchain.RegisterIndexer(ti, "") + blockchain.indexServers.servers[0].testSuspendHookCh = testSuspendHookCh + }) + ti.expEvent(testIndexerEvent{ev: "SetHistoryCutoff", blockNumber: 0}) + ti.expEvent(testIndexerEvent{ev: "SetFinalized", blockNumber: 0}) + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: 0, blockHash: chain[0].Hash()}) + ti.status(testIndexerStatus{ready: true}) + expDone() + + insert(chain[1:101]) + waitSuspend() + ti.expEvent(testIndexerEvent{ev: "Suspended"}) + for i := uint64(1); i <= 100; i++ { + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: i, blockHash: chain[i].Hash()}) + ti.status(testIndexerStatus{ready: true}) + } + expDone() + + run(blockchain.Stop) + ti.expEvent(testIndexerEvent{ev: "Stop"}) + expDone() + + blockchain, _ = NewBlockChain(db, gspec, ethash.NewFaker(), DefaultConfig()) + run(func() { + blockchain.RegisterIndexer(ti, "") + blockchain.indexServers.servers[0].testSuspendHookCh = testSuspendHookCh + }) + ti.expEvent(testIndexerEvent{ev: "SetHistoryCutoff", blockNumber: 0}) + ti.expEvent(testIndexerEvent{ev: "SetFinalized", blockNumber: 0}) + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: 100, blockHash: chain[100].Hash()}) + ti.status(testIndexerStatus{ready: true, needBlocks: common.NewRange[uint64](0, 100)}) + expDone() + // request entire chain as historical range, add a new block in the middle and check suspend mechanism + for i := uint64(0); i <= 49; i++ { + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: i, blockHash: chain[i].Hash()}) + ti.status(testIndexerStatus{ready: true, needBlocks: common.NewRange[uint64](i+1, 99-i)}) + } + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: 50, blockHash: chain[50].Hash()}) + insert(chain[101:102]) + waitSuspend() + ti.status(testIndexerStatus{ready: true, needBlocks: common.NewRange[uint64](51, 49)}) + ti.expEvent(testIndexerEvent{ev: "Suspended"}) + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: 101, blockHash: chain[101].Hash()}) + ti.status(testIndexerStatus{ready: true, needBlocks: common.NewRange[uint64](51, 50)}) + expDone() + for i := uint64(51); i <= 100; i++ { + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: i, blockHash: chain[i].Hash()}) + ti.status(testIndexerStatus{ready: true, needBlocks: common.NewRange[uint64](i+1, 100-i)}) + } + + run(func() { + blockchain.SetHead(80) + }) + ti.expEvent(testIndexerEvent{ev: "Revert", blockNumber: 80}) + expDone() + chain = chain[:81] + blocks, _ = GenerateChain(gspec.Config, chain[80], ethash.NewFaker(), db, 45, func(i int, gen *BlockGen) {}) + chain = append(chain, blocks...) + insert(chain[81:121]) + waitSuspend() + ti.expEvent(testIndexerEvent{ev: "Suspended"}) + for i := uint64(81); i <= 120; i++ { + ti.expEvent(testIndexerEvent{ev: "AddBlockData", blockNumber: i, blockHash: chain[i].Hash()}) + ti.status(testIndexerStatus{ready: true}) + } + expDone() + + run(blockchain.Stop) + ti.expEvent(testIndexerEvent{ev: "Stop"}) + expDone() +} + +type testIndexer struct { + t *testing.T + eventCh chan testIndexerEvent + statusCh chan testIndexerStatus +} + +type testIndexerEvent struct { + ev string + blockNumber uint64 + blockHash common.Hash +} + +type testIndexerStatus struct { + ready bool + needBlocks common.Range[uint64] +} + +func (ti *testIndexer) expEvent(exp testIndexerEvent) { + var got testIndexerEvent + select { + case got = <-ti.eventCh: + case <-time.After(time.Second * 5): + } + if got != exp { + ti.t.Fatalf("Wrong indexer event received (expected: %v, got: %v)", exp, got) + } +} + +func (ti *testIndexer) status(status testIndexerStatus) { + ti.statusCh <- status +} + +func (ti *testIndexer) AddBlockData(header *types.Header, receipts types.Receipts) (ready bool, needBlocks common.Range[uint64]) { + ti.eventCh <- testIndexerEvent{ev: "AddBlockData", blockNumber: header.Number.Uint64(), blockHash: header.Hash()} + status := <-ti.statusCh + return status.ready, status.needBlocks +} + +func (ti *testIndexer) Revert(blockNumber uint64) { + ti.eventCh <- testIndexerEvent{ev: "Revert", blockNumber: blockNumber} +} + +func (ti *testIndexer) Status() (ready bool, needBlocks common.Range[uint64]) { + ti.eventCh <- testIndexerEvent{ev: "Status"} + status := <-ti.statusCh + return status.ready, status.needBlocks +} + +func (ti *testIndexer) SetHistoryCutoff(blockNumber uint64) { + ti.eventCh <- testIndexerEvent{ev: "SetHistoryCutoff", blockNumber: blockNumber} +} + +func (ti *testIndexer) SetFinalized(blockNumber uint64) { + ti.eventCh <- testIndexerEvent{ev: "SetFinalized", blockNumber: blockNumber} +} + +func (ti *testIndexer) Suspended() { + ti.eventCh <- testIndexerEvent{ev: "Suspended"} +} + +func (ti *testIndexer) Stop() { + ti.eventCh <- testIndexerEvent{ev: "Stop"} +}