various indexer fixes

ethpandaops · Jul 31, 2024 · bf484d5 · bf484d5
1 parent a2ef656
commit bf484d5
Show file tree

Hide file tree

Showing 5 changed files with 28 additions and 22 deletions.
diff --git a/indexer/beacon/client.go b/indexer/beacon/client.go
@@ -187,36 +187,29 @@ func (c *Client) processHeadEvent(headEvent *v1.HeadEvent) error {
 	if !bytes.Equal(dependentRoot[:], consensus.NullRoot[:]) {
 		block.dependentRoot = &dependentRoot
 
-		dependentBlock := c.indexer.blockCache.getBlockByRoot(dependentRoot)
+		dependentBlock = c.indexer.blockCache.getBlockByRoot(dependentRoot)
 		if dependentBlock == nil {
 			c.logger.Warnf("dependent block (%v) not found after backfilling", dependentRoot.String())
 		}
+	} else {
+		dependentBlock = c.indexer.blockCache.getDependentBlock(chainState, block, c)
 	}
 
 	currentBlock := block
 	minInMemorySlot := c.indexer.getMinInMemorySlot()
 	for {
-		if bytes.Equal(dependentRoot[:], consensus.NullRoot[:]) || dependentBlock == nil {
-			dependentBlock = c.indexer.blockCache.getDependentBlock(chainState, currentBlock, c)
-			if dependentBlock != nil {
-				dependentRoot = dependentBlock.Root
-			} else {
-				dependentRoot = consensus.NullRoot
-			}
-		}
-
-		if !bytes.Equal(dependentRoot[:], consensus.NullRoot[:]) && currentBlock.Slot >= minInMemorySlot {
+		if dependentBlock != nil && currentBlock.Slot >= minInMemorySlot {
 			// ensure epoch stats are in loading queue
-			epochStats, _ := c.indexer.epochCache.createOrGetEpochStats(chainState.EpochOfSlot(currentBlock.Slot), dependentRoot)
+			epochStats, _ := c.indexer.epochCache.createOrGetEpochStats(chainState.EpochOfSlot(currentBlock.Slot), dependentBlock.Root)
 			if !epochStats.addRequestedBy(c) {
 				break
 			}
-
-			currentBlock = dependentBlock
-			dependentBlock = nil
 		} else {
 			break
 		}
+
+		currentBlock = dependentBlock
+		dependentBlock = c.indexer.blockCache.getDependentBlock(chainState, currentBlock, c)
 	}
 
 	c.headRoot = block.Root

diff --git a/indexer/beacon/epochstats.go b/indexer/beacon/epochstats.go
@@ -369,10 +369,18 @@ func (es *EpochStats) processState(indexer *Indexer) {
 }
 
 // GetValues returns the EpochStats values.
-func (es *EpochStats) GetValues() *EpochStatsValues {
+func (es *EpochStats) GetValues(chainState *consensus.ChainState) *EpochStatsValues {
 	if es == nil {
 		return nil
 	}
 
-	return es.values
+	if es.values != nil {
+		return es.values
+	}
+
+	if es.packedValues != nil {
+		return es.getUnpackedValues(chainState)
+	}
+
+	return nil
 }
diff --git a/indexer/beacon/indexer.go b/indexer/beacon/indexer.go
@@ -161,10 +161,15 @@ func (indexer *Indexer) StartIndexer() {
 		}
 
 		restoredEpochStats++
-		if dbDuty.Epoch < uint64(indexer.lastPrunedEpoch) {
-			epochStats.packValues()
+		if dbDuty.Epoch >= uint64(indexer.lastPrunedEpoch) {
+			epochStats.unpackValues(chainState)
 		}
 	})
+	if err != nil {
+		indexer.logger.WithError(err).Errorf("failed restoring unfinalized epoch stats from DB")
+	} else {
+		indexer.logger.Infof("restored %v unfinalized epoch stats from DB", restoredEpochStats)
+	}
 
 	// prefill block cache with all unfinalized blocks from db
 	restoredBlockCount := 0

diff --git a/indexer/beacon/processing.go b/indexer/beacon/processing.go
@@ -243,7 +243,7 @@ func (indexer *Indexer) finalizeEpoch(epoch phase0.Epoch, justifiedRoot phase0.R
 		return fmt.Errorf("missing epoch stats for epoch %v", epoch)
 	}
 
-	epochStatsValues := epochStats.GetValues()
+	epochStatsValues := epochStats.GetValues(chainState)
 	if epochStatsValues == nil {
 		return fmt.Errorf("missing epoch stats values for epoch %v", epoch)
 	}

diff --git a/indexer/beacon/writedb.go b/indexer/beacon/writedb.go
@@ -24,7 +24,7 @@ func newDbWriter(indexer *Indexer) *dbWriter {
 
 func (dbw *dbWriter) persistMissedSlots(tx *sqlx.Tx, epoch phase0.Epoch, blocks []*Block, epochStats *EpochStats) error {
 	chainState := dbw.indexer.consensusPool.GetChainState()
-	epochStatsValues := epochStats.GetValues()
+	epochStatsValues := epochStats.GetValues(chainState)
 
 	// insert missed slots
 	firstSlot := chainState.EpochStartSlot(epoch)
@@ -255,7 +255,7 @@ func (dbw *dbWriter) buildDbBlock(block *Block, epochStats *EpochStats, override
 
 func (dbw *dbWriter) buildDbEpoch(epoch phase0.Epoch, blocks []*Block, epochStats *EpochStats, epochVotes *EpochVotes, blockFn func(block *Block, depositIndex *uint64)) *dbtypes.Epoch {
 	chainState := dbw.indexer.consensusPool.GetChainState()
-	epochStatsValues := epochStats.GetValues()
+	epochStatsValues := epochStats.GetValues(chainState)
 
 	// insert missed slots
 	firstSlot := chainState.EpochStartSlot(epoch)