debug committer

Author: iuwqyir

internal/orchestrator/committer.go

@@ -7,6 +7,7 @@ import (
 	"sort"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/rs/zerolog/log"
 	config "github.com/thirdweb-dev/indexer/configs"
 	"github.com/thirdweb-dev/indexer/internal/common"
@@ -57,7 +58,8 @@ func (c *Committer) Start(ctx context.Context) {
 			log.Info().Msg("Committer shutting down")
 			return
 		case <-ticker.C:
-			blockDataToCommit, err := c.getSequentialBlockDataToCommit()
+			commitID := uuid.New().String()
+			blockDataToCommit, err := c.getSequentialBlockDataToCommit(commitID)
 			if err != nil {
 				log.Error().Err(err).Msg("Error getting block data to commit")
 				continue
@@ -66,16 +68,16 @@ func (c *Committer) Start(ctx context.Context) {
 				log.Debug().Msg("No block data to commit")
 				continue
 			}
-			if err := c.commit(blockDataToCommit); err != nil {
+			if err := c.commit(commitID, blockDataToCommit); err != nil {
 				log.Error().Err(err).Msg("Error committing blocks")
 			}
 		}
 	}
 }
 
-func (c *Committer) getBlockNumbersToCommit() ([]*big.Int, error) {
+func (c *Committer) getBlockNumbersToCommit(commitID string) ([]*big.Int, error) {
 	latestCommittedBlockNumber, err := c.storage.MainStorage.GetMaxBlockNumber(c.rpc.GetChainID())
-	log.Info().Msgf("Committer found this max block number in main storage: %s", latestCommittedBlockNumber.String())
+	log.Info().Str("commitID", commitID).Msgf("Committer found this max block number in main storage: %s", latestCommittedBlockNumber.String())
 	if err != nil {
 		return nil, err
 	}
@@ -97,8 +99,8 @@ func (c *Committer) getBlockNumbersToCommit() ([]*big.Int, error) {
 	return blockNumbers, nil
 }
 
-func (c *Committer) getSequentialBlockDataToCommit() (*[]common.BlockData, error) {
-	blocksToCommit, err := c.getBlockNumbersToCommit()
+func (c *Committer) getSequentialBlockDataToCommit(commitID string) (*[]common.BlockData, error) {
+	blocksToCommit, err := c.getBlockNumbersToCommit(commitID)
 	if err != nil {
 		return nil, fmt.Errorf("error determining blocks to commit: %v", err)
 	}
@@ -111,8 +113,8 @@ func (c *Committer) getSequentialBlockDataToCommit() (*[]common.BlockData, error
 		return nil, fmt.Errorf("error fetching blocks to commit: %v", err)
 	}
 	if blocksData == nil || len(*blocksData) == 0 {
-		log.Warn().Msgf("Committer didn't find the following range in staging: %v - %v", blocksToCommit[0].Int64(), blocksToCommit[len(blocksToCommit)-1].Int64())
-		c.handleMissingStagingData(blocksToCommit)
+		log.Warn().Str("commitID", commitID).Msgf("Committer didn't find the following range in staging: %v - %v", blocksToCommit[0].Int64(), blocksToCommit[len(blocksToCommit)-1].Int64())
+		c.handleMissingStagingData(commitID, blocksToCommit)
 		return nil, nil
 	}
 
@@ -136,7 +138,7 @@ func (c *Committer) getSequentialBlockDataToCommit() (*[]common.BlockData, error
 		}
 		if (*blocksData)[i].Block.Number.Cmp(expectedBlockNumber) != 0 {
 			// Note: Gap detected, stop here
-			log.Warn().Msgf("Gap detected at block %s, committing until %s", expectedBlockNumber.String(), (*blocksData)[i-1].Block.Number.String())
+			log.Warn().Str("commitID", commitID).Msgf("Gap detected at block %s, committing until %s", expectedBlockNumber.String(), (*blocksData)[i-1].Block.Number.String())
 			// increment the a gap counter in prometheus
 			metrics.GapCounter.Inc()
 			// record the first missed block number in prometheus
@@ -150,22 +152,24 @@ func (c *Committer) getSequentialBlockDataToCommit() (*[]common.BlockData, error
 	return &sequentialBlockData, nil
 }
 
-func (c *Committer) commit(blockData *[]common.BlockData) error {
+func (c *Committer) commit(commitID string, blockData *[]common.BlockData) error {
 	blockNumbers := make([]*big.Int, len(*blockData))
 	for i, block := range *blockData {
 		blockNumbers[i] = block.Block.Number
 	}
-	log.Debug().Msgf("Committing %d blocks", len(blockNumbers))
+	log.Debug().Str("commitID", commitID).Msgf("Committing %d blocks: %v", len(blockNumbers), blockNumbers)
 
 	// TODO if next parts (saving or deleting) fail, we'll have to do a rollback
 	if err := c.storage.MainStorage.InsertBlockData(blockData); err != nil {
 		log.Error().Err(err).Msgf("Failed to commit blocks: %v", blockNumbers)
 		return fmt.Errorf("error saving data to main storage: %v", err)
 	}
+	log.Debug().Str("commitID", commitID).Msgf("Committer inserted blocks: %v", blockNumbers)
 
 	if err := c.storage.StagingStorage.DeleteStagingData(blockData); err != nil {
 		return fmt.Errorf("error deleting data from staging storage: %v", err)
 	}
+	log.Debug().Str("commitID", commitID).Msgf("Committer deleted staging data for blocks: %v", blockNumbers)
 
 	// Update metrics for successful commits
 	metrics.SuccessfulCommits.Add(float64(len(*blockData)))
@@ -199,24 +203,24 @@ func (c *Committer) handleGap(expectedStartBlockNumber *big.Int, actualFirstBloc
 	return fmt.Errorf("first block number (%s) in commit batch does not match expected (%s)", actualFirstBlock.Number.String(), expectedStartBlockNumber.String())
 }
 
-func (c *Committer) handleMissingStagingData(blocksToCommit []*big.Int) {
+func (c *Committer) handleMissingStagingData(commitID string, blocksToCommit []*big.Int) {
 	// Checks if there are any blocks in staging after the current range end
 	lastStagedBlockNumber, err := c.storage.StagingStorage.GetLastStagedBlockNumber(c.rpc.GetChainID(), blocksToCommit[len(blocksToCommit)-1], big.NewInt(0))
 	if err != nil {
 		log.Error().Err(err).Msg("Error checking staged data for missing range")
 		return
 	}
 	if lastStagedBlockNumber == nil || lastStagedBlockNumber.Sign() <= 0 {
-		log.Debug().Msgf("Committer is caught up with staging. No need to poll for missing blocks.")
+		log.Debug().Str("commitID", commitID).Msgf("Committer is caught up with staging. No need to poll for missing blocks.")
 		return
 	}
-	log.Debug().Msgf("Detected missing blocks in staging data starting from %s.", blocksToCommit[0].String())
+	log.Debug().Str("commitID", commitID).Msgf("Detected missing blocks in staging data starting from %s.", blocksToCommit[0].String())
 
 	poller := NewBoundlessPoller(c.rpc, c.storage)
 	blocksToPoll := blocksToCommit
 	if len(blocksToCommit) > int(poller.blocksPerPoll) {
 		blocksToPoll = blocksToCommit[:int(poller.blocksPerPoll)]
 	}
 	poller.Poll(blocksToPoll)
-	log.Debug().Msgf("Polled %d blocks due to committer detecting them as missing. Range: %s - %s", len(blocksToPoll), blocksToPoll[0].String(), blocksToPoll[len(blocksToPoll)-1].String())
+	log.Debug().Str("commitID", commitID).Msgf("Polled %d blocks due to committer detecting them as missing. Range: %s - %s", len(blocksToPoll), blocksToPoll[0].String(), blocksToPoll[len(blocksToPoll)-1].String())
 }