toolbar/vote_reward/synchron/block_keeper.go

   1 package synchron
   2
   3 import (
   4         "encoding/hex"
   5
   6         "github.com/jinzhu/gorm"
   7         log "github.com/sirupsen/logrus"
   8
   9         "github.com/bytom/vapor/errors"
  10         "github.com/bytom/vapor/protocol/bc/types"
  11         "github.com/bytom/vapor/toolbar/apinode"
  12         "github.com/bytom/vapor/toolbar/common"
  13         "github.com/bytom/vapor/toolbar/vote_reward/config"
  14         "github.com/bytom/vapor/toolbar/vote_reward/database/orm"
  15 )
  16
  17 var ErrInconsistentDB = errors.New("inconsistent db status")
  18
  19 type ChainKeeper struct {
  20         db           *gorm.DB
  21         node         *apinode.Node
  22         targetHeight uint64
  23 }
  24
  25 func NewChainKeeper(db *gorm.DB, cfg *config.Config, targetHeight uint64) (*ChainKeeper, error) {
  26         keeper := &ChainKeeper{
  27                 db:           db,
  28                 node:         apinode.NewNode(cfg.NodeIP),
  29                 targetHeight: targetHeight,
  30         }
  31
  32         chainStatus := &orm.ChainStatus{}
  33         if err := db.First(chainStatus).Error; err == nil {
  34                 return keeper, nil
  35         } else if err != gorm.ErrRecordNotFound {
  36                 return nil, errors.Wrap(err, "fail on get chainStatus")
  37         }
  38
  39         if err := keeper.initBlockState(); err != nil {
  40                 return nil, errors.Wrap(err, "fail on init chainStatus")
  41         }
  42         return keeper, nil
  43 }
  44
  45 func (c *ChainKeeper) SyncBlock() error {
  46         for {
  47                 chainStatus := &orm.ChainStatus{}
  48                 if err := c.db.First(chainStatus).Error; err != nil {
  49                         return errors.Wrap(err, "fail on syncBlock query chainStatus")
  50                 }
  51
  52                 if chainStatus.BlockHeight >= c.targetHeight {
  53                         break
  54                 }
  55
  56                 dbTX := c.db.Begin()
  57                 if err := c.syncChainStatus(dbTX, chainStatus); err != nil {
  58                         dbTX.Rollback()
  59                         return err
  60                 }
  61
  62                 if err := dbTX.Commit().Error; err != nil {
  63                         return err
  64                 }
  65         }
  66         return nil
  67 }
  68
  69 func (c *ChainKeeper) syncChainStatus(db *gorm.DB, chainStatus *orm.ChainStatus) error {
  70         nextBlock, err := c.node.GetBlockByHeight(chainStatus.BlockHeight + 1)
  71         if err != nil {
  72                 return err
  73         }
  74
  75         // Normal case, the previous hash of next block equals to the hash of current block,
  76         // just sync to database directly.
  77         if nextBlock.PreviousBlockHash.String() == chainStatus.BlockHash {
  78                 return c.AttachBlock(db, chainStatus, nextBlock)
  79         }
  80
  81         log.WithField("block height", chainStatus.BlockHeight).Debug("the prev hash of remote is not equals the hash of current best block, must rollback")
  82         currentBlock, err := c.node.GetBlockByHash(chainStatus.BlockHash)
  83         if err != nil {
  84                 return err
  85         }
  86
  87         preBlock, err := c.node.GetBlockByHash(currentBlock.PreviousBlockHash.String())
  88         if err != nil {
  89                 return err
  90         }
  91
  92         return c.DetachBlock(db, chainStatus, preBlock)
  93 }
  94
  95 func (c *ChainKeeper) AttachBlock(db *gorm.DB, chainStatus *orm.ChainStatus, block *types.Block) error {
  96         for _, tx := range block.Transactions {
  97                 for _, input := range tx.Inputs {
  98                         if input.TypedInput.InputType() != types.VetoInputType {
  99                                 continue
 100                         }
 101
 102                         outputID, err := input.SpentOutputID()
 103                         if err != nil {
 104                                 return err
 105                         }
 106
 107                         result := db.Model(&orm.Utxo{}).Where(&orm.Utxo{OutputID: outputID.String()}).Update("veto_height", block.Height)
 108                         if err := result.Error; err != nil {
 109                                 return err
 110                         } else if result.RowsAffected != 1 {
 111                                 return ErrInconsistentDB
 112                         }
 113                 }
 114
 115                 for i, output := range tx.Outputs {
 116                         voteOutput, ok := output.TypedOutput.(*types.VoteOutput)
 117                         if !ok {
 118                                 continue
 119                         }
 120
 121                         utxo := &orm.Utxo{
 122                                 Xpub:        hex.EncodeToString(voteOutput.Vote),
 123                                 VoteAddress: common.GetAddressFromControlProgram(voteOutput.ControlProgram),
 124                                 VoteHeight:  block.Height,
 125                                 VoteNum:     voteOutput.Amount,
 126                                 OutputID:    tx.OutputID(i).String(),
 127                         }
 128
 129                         if err := db.Save(utxo).Error; err != nil {
 130                                 return err
 131                         }
 132                 }
 133         }
 134
 135         return c.updateChainStatus(db, chainStatus, block)
 136 }
 137
 138 func (c *ChainKeeper) DetachBlock(db *gorm.DB, chainStatus *orm.ChainStatus, block *types.Block) error {
 139         if err := db.Where(&orm.Utxo{VoteHeight: chainStatus.BlockHeight}).Delete(&orm.Utxo{}).Error; err != nil {
 140                 return err
 141         }
 142
 143         if err := db.Model(&orm.Utxo{}).Where(&orm.Utxo{VetoHeight: chainStatus.BlockHeight}).Update("veto_height", 0).Error; err != nil {
 144                 return err
 145         }
 146
 147         return c.updateChainStatus(db, chainStatus, block)
 148 }
 149
 150 func (c *ChainKeeper) initBlockState() error {
 151         block, err := c.node.GetBlockByHeight(0)
 152         if err != nil {
 153                 return errors.Wrap(err, "fail on get genenis block")
 154         }
 155
 156         blockHash := block.Hash()
 157         chainStatus := &orm.ChainStatus{
 158                 BlockHeight: block.Height,
 159                 BlockHash:   blockHash.String(),
 160         }
 161         return c.db.Save(chainStatus).Error
 162 }
 163
 164 func (c *ChainKeeper) updateChainStatus(db *gorm.DB, chainStatus *orm.ChainStatus, block *types.Block) error {
 165         blockHash := block.Hash()
 166         result := db.Model(&orm.ChainStatus{}).Where(chainStatus).Updates(&orm.ChainStatus{
 167                 BlockHeight: block.Height,
 168                 BlockHash:   blockHash.String(),
 169         })
 170         if err := result.Error; err != nil {
 171                 return err
 172         } else if result.RowsAffected != 1 {
 173                 return ErrInconsistentDB
 174         }
 175         return nil
 176 }