core: add Indexer interface and index server logic

Author: zsfelfoldi

core/blockchain.go

@@ -303,6 +303,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.
@@ -523,9 +524,15 @@ func NewBlockChain(db ethdb.Database, genesis *Genesis, engine consensus.Engine,
 	if bc.cfg.TxLookupLimit >= 0 {
 		bc.txIndexer = newTxIndexer(uint64(bc.cfg.TxLookupLimit), bc)
 	}
+	bc.indexServers.init(bc)
 	return bc, nil
 }
 
+// RegisterIndexer registers a new indexer to the chain.
+func (bc *BlockChain) RegisterIndexer(indexer Indexer) {
+	bc.indexServers.register(indexer)
+}
+
 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.
@@ -638,6 +645,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()))
@@ -685,6 +693,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:
@@ -706,6 +715,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:
@@ -768,9 +778,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)
 	}
 }
 
@@ -1133,6 +1145,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
@@ -1149,6 +1162,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(), true)
 
 	// Last update all in-memory chain markers
 	bc.genesisBlock = genesis
@@ -1261,6 +1275,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.
@@ -1562,6 +1577,7 @@ func (bc *BlockChain) writeKnownBlock(block *types.Block) error {
 		}
 	}
 	bc.writeHeadBlock(block)
+	bc.indexServers.broadcast(block.Header(), true)
 	return nil
 }
 
@@ -1577,7 +1593,10 @@ 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)
+	bc.receiptsCache.Add(blockHash, receipts)
 	rawdb.WritePreimages(blockBatch, statedb.Preimages())
 	if err := blockBatch.Write(); err != nil {
 		log.Crit("Failed to write block into disk", "err", err)
@@ -1664,6 +1683,7 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
 
 	// Set new head.
 	bc.writeHeadBlock(block)
+	bc.indexServers.broadcast(block.Header(), true)
 
 	bc.chainFeed.Send(ChainEvent{Header: block.Header()})
 	if len(logs) > 0 {
@@ -2373,6 +2393,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
@@ -2470,6 +2491,7 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Header) error
 		}
 		// Update the head block
 		bc.writeHeadBlock(block)
+		bc.indexServers.broadcast(block.Header(), false)
 	}
 	if len(rebirthLogs) > 0 {
 		bc.logsFeed.Send(rebirthLogs)
@@ -2545,6 +2567,7 @@ func (bc *BlockChain) SetCanonical(head *types.Block) (common.Hash, error) {
 		}
 	}
 	bc.writeHeadBlock(head)
+	bc.indexServers.broadcast(head.Header(), true)
 
 	// Emit events
 	logs := bc.collectLogs(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