core: add Indexer interface and index server logic

Author: zsfelfoldi

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)

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