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"}
+}