diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 7cd724994..e4fd04cd3 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -1127,7 +1127,7 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server. } StateDiffBackfillMaxHeadGap = &cli.Uint64Flag{ Name: "statediff.backfillmaxheadgap", - Usage: "The maximum gap between the startup statediff and startup head positions that can be backfilled.", + Usage: "The maximum gap between the current statediff and head positions that can be backfilled.", Value: 7200, } ) diff --git a/statediff/service.go b/statediff/service.go index 64a7b0b95..9839cd048 100644 --- a/statediff/service.go +++ b/statediff/service.go @@ -170,6 +170,12 @@ type statusSubscription struct { quitChan chan<- bool } +// Utility type for showing the relative positions of the blockchain and the statediff indexer. +type servicePosition struct { + chainBlockNumber uint64 + indexerBlockNumber uint64 +} + // BlockCache caches the last block for safe access from different service loops type BlockCache struct { sync.Mutex @@ -322,32 +328,59 @@ func (lbc *BlockCache) getParentBlock(currentBlock *types.Block, bc blockChain) } type workerParams struct { - chainEventCh <-chan core.ChainEvent - wg *sync.WaitGroup - id uint + blockCh <-chan *types.Block + wg *sync.WaitGroup + id uint } func (sds *Service) WriteLoop(chainEventCh chan core.ChainEvent) { + initialPos := sds.currentPosition() + log.Info( + "WriteLoop: initial positions", + "chain", initialPos.chainBlockNumber, + "indexer", initialPos.indexerBlockNumber, + ) chainEventSub := sds.BlockChain.SubscribeChainEvent(chainEventCh) defer chainEventSub.Unsubscribe() errCh := chainEventSub.Err() var wg sync.WaitGroup // Process metrics for chain events, then forward to workers - chainEventFwd := make(chan core.ChainEvent, chainEventChanSize) + blockFwd := make(chan *types.Block, chainEventChanSize) wg.Add(1) go func() { defer wg.Done() for { select { case chainEvent := <-chainEventCh: - lastHeight := defaultStatediffMetrics.lastEventHeight.Value() - nextHeight := int64(chainEvent.Block.Number().Uint64()) - if nextHeight-lastHeight != 1 { - log.Warn("Statediffing service received block out-of-order", "next height", nextHeight, "last height", lastHeight) + lastHeight := uint64(defaultStatediffMetrics.lastEventHeight.Value()) + if lastHeight == 0 { + lastHeight = initialPos.indexerBlockNumber + } + nextHeight := chainEvent.Block.Number().Uint64() + if nextHeight > lastHeight { + distance := nextHeight - lastHeight + if distance == 1 { + log.Info("WriteLoop: received expected block", "block height", nextHeight, "last height", lastHeight) + blockFwd <- chainEvent.Block + defaultStatediffMetrics.lastEventHeight.Update(int64(nextHeight)) + } else { + log.Warn("WriteLoop: received unexpected block from the future", "block height", nextHeight, "last height", lastHeight) + if distance <= sds.backfillMaxHeadGap { + for i := lastHeight + 1; i < nextHeight; i++ { + log.Info("WriteLoop: backfilling gap to head", "block", i, "block height", nextHeight, "last height", lastHeight) + blockFwd <- sds.BlockChain.GetBlockByNumber(i) + } + } else { + log.Warn("WriteLoop: gap to head too large to backfill", "block height", nextHeight, "last height", lastHeight, "gap", distance) + } + blockFwd <- chainEvent.Block + defaultStatediffMetrics.lastEventHeight.Update(int64(nextHeight)) + } + } else { + log.Warn("WriteLoop: received unexpected block from the past", "block height", nextHeight, "last height", lastHeight) + blockFwd <- chainEvent.Block } - defaultStatediffMetrics.lastEventHeight.Update(nextHeight) defaultStatediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh))) - chainEventFwd <- chainEvent case err := <-errCh: log.Error("Error from chain event subscription", "error", err) close(sds.QuitChan) @@ -367,7 +400,7 @@ func (sds *Service) WriteLoop(chainEventCh chan core.ChainEvent) { }() wg.Add(int(sds.numWorkers)) for worker := uint(0); worker < sds.numWorkers; worker++ { - params := workerParams{chainEventCh: chainEventFwd, wg: &wg, id: worker} + params := workerParams{blockCh: blockFwd, wg: &wg, id: worker} go sds.writeLoopWorker(params) } wg.Wait() @@ -392,9 +425,9 @@ func (sds *Service) writeLoopWorker(params workerParams) { for { select { //Notify chain event channel of events - case chainEvent := <-params.chainEventCh: + case chainEvent := <-params.blockCh: log.Debug("WriteLoop(): chain event received", "event", chainEvent) - currentBlock := chainEvent.Block + currentBlock := chainEvent parentBlock := sds.BlockCache.getParentBlock(currentBlock, sds.BlockChain) if parentBlock == nil { log.Error("Parent block is nil, skipping this block", "block height", currentBlock.Number()) @@ -438,6 +471,8 @@ func (sds *Service) Loop(chainEventCh chan core.ChainEvent) { select { //Notify chain event channel of events case chainEvent := <-chainEventCh: + // TODO: Do we need to track the last streamed block as we do for the WriteLoop so that we can detect + // and plug any gaps in the events? If not, we risk presenting an incomplete record. defaultStatediffMetrics.serviceLoopChannelLen.Update(int64(len(chainEventCh))) log.Debug("Loop(): chain event received", "event", chainEvent) // if we don't have any subscribers, do not process a statediff @@ -1114,57 +1149,24 @@ func MapWatchAddressArgsToAddresses(args []types2.WatchAddressArg) ([]common.Add // Backfill is executed on startup to make sure there are no gaps in the recent past when tracking head. func (sds *Service) Backfill() { - chainBlock := sds.BlockChain.CurrentBlock() - if nil == chainBlock { - log.Info("Backfill: No previous chain block, nothing to backfill.") - return - } - - chainBlockNumber := chainBlock.Number.Uint64() - if chainBlockNumber == 0 { + pos := sds.currentPosition() + if pos.chainBlockNumber == 0 { log.Info("Backfill: At start of chain, nothing to backfill.") return } - indexerBlock, err := sds.indexer.CurrentBlock() - if nil == indexerBlock { - log.Info("Backfill: No previous indexer block, nothing to backfill.") - return - } - if nil != err { - log.Error("Backfill error: " + err.Error()) - return - } - - indexerBlockNumber, err := strconv.ParseUint(indexerBlock.BlockNumber, 10, 64) - if nil != err { - log.Error("Backfill error: " + err.Error()) - return - } - - headGap := chainBlockNumber - indexerBlockNumber log.Info( "Backfill: initial positions", - "chain", chainBlockNumber, - "indexer", indexerBlockNumber, - "headGap", headGap, + "chain", pos.chainBlockNumber, + "indexer", pos.indexerBlockNumber, ) - if sds.backfillMaxHeadGap > 0 && headGap > 0 { - if headGap < sds.backfillMaxHeadGap { - sds.backfillHeadGap(indexerBlockNumber, chainBlockNumber) - log.Info("Backfill: all workers done filling headGap.") - } else { - log.Error("Backfill: headGap too large to fill.") - } - } - if sds.backfillCheckPastBlocks > 0 { var gapCheckBeginNumber uint64 = 0 - if indexerBlockNumber > sds.backfillCheckPastBlocks { - gapCheckBeginNumber = indexerBlockNumber - sds.backfillCheckPastBlocks + if pos.indexerBlockNumber > sds.backfillCheckPastBlocks { + gapCheckBeginNumber = pos.indexerBlockNumber - sds.backfillCheckPastBlocks } - blockGaps, err := sds.indexer.DetectGaps(gapCheckBeginNumber, chainBlockNumber) + blockGaps, err := sds.indexer.DetectGaps(gapCheckBeginNumber, pos.chainBlockNumber) if nil != err { log.Error("Backfill error: " + err.Error()) return @@ -1172,52 +1174,15 @@ func (sds *Service) Backfill() { if nil != blockGaps && len(blockGaps) > 0 { gapsMsg, _ := json.Marshal(blockGaps) - log.Info("Backfill: detected gaps in range", "begin", gapCheckBeginNumber, "end", chainBlockNumber, "gaps", string(gapsMsg)) + log.Info("Backfill: detected gaps in range", "begin", gapCheckBeginNumber, "end", pos.chainBlockNumber, "gaps", string(gapsMsg)) sds.backfillDetectedGaps(blockGaps) - log.Info("Backfill: done processing detected gaps in range", "begin", gapCheckBeginNumber, "end", chainBlockNumber, "gaps", string(gapsMsg)) + log.Info("Backfill: done processing detected gaps in range", "begin", gapCheckBeginNumber, "end", pos.chainBlockNumber, "gaps", string(gapsMsg)) } else { - log.Info("Backfill: no gaps detected in range", "begin", gapCheckBeginNumber, "end", chainBlockNumber) + log.Info("Backfill: no gaps detected in range", "begin", gapCheckBeginNumber, "end", pos.chainBlockNumber) } } } -// backfillHeadGap fills in any gap between the statediff position and the chain position at startup. -// A gap can be created in this way if there is some problem in statediffing (eg, DB connectivity is lost, -// while the chain keeps syncing), if the process is terminated with a statediff in-flight, etc. -func (sds *Service) backfillHeadGap(indexerBlockNumber uint64, chainBlockNumber uint64) { - var ch = make(chan uint64) - var wg sync.WaitGroup - for i := uint(0); i < sds.numWorkers; i++ { - wg.Add(1) - go func(w uint) { - defer wg.Done() - for { - select { - case num, ok := <-ch: - if !ok { - log.Info("Backfill: headGap done", "worker", w) - return - } - log.Info("Backfill: backfilling head gap", "block", num, "worker", w) - err := sds.writeStateDiffAt(num, writeLoopParams.Params) - if err != nil { - log.Error("Backfill error: " + err.Error()) - } - case <-sds.QuitChan: - log.Info("Backfill: quitting before finish", "worker", w) - return - } - } - }(i) - } - - for bn := indexerBlockNumber + 1; bn <= chainBlockNumber; bn++ { - ch <- bn - } - close(ch) - wg.Wait() -} - // backfillDetectedGaps fills gaps which have occurred in the recent past. These gaps can happen because of // transient errors, such as DB errors that are later corrected (so head statediffing continues, but with a hole) // a missed ChainEvent (happens sometimes when debugging), or if the process is terminated when an earlier block @@ -1257,3 +1222,24 @@ func (sds *Service) backfillDetectedGaps(blockGaps []*interfaces.BlockGap) { close(ch) wg.Wait() } + +// currentPosition returns the current block height for both the BlockChain and the statediff indexer. +func (sds *Service) currentPosition() servicePosition { + ret := servicePosition{} + chainBlock := sds.BlockChain.CurrentBlock() + if nil != chainBlock { + ret.chainBlockNumber = chainBlock.Number.Uint64() + } + + indexerBlock, _ := sds.indexer.CurrentBlock() + if nil != indexerBlock { + indexerBlockNumber, err := strconv.ParseUint(indexerBlock.BlockNumber, 10, 64) + if nil == err { + ret.indexerBlockNumber = indexerBlockNumber + } else { + log.Error("Error parsing indexer block number", "block", indexerBlock.BlockNumber) + } + } + + return ret +}