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/vapor/errors"
  10         "github.com/vapor/protocol/bc/types"
  11         "github.com/vapor/toolbar/apinode"
  12         "github.com/vapor/toolbar/common"
  13         "github.com/vapor/toolbar/vote_reward/config"
  14         "github.com/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         return c.DetachBlock(db, chainStatus, currentBlock)
  88 }
  89
  90 func (c *ChainKeeper) AttachBlock(db *gorm.DB, chainStatus *orm.ChainStatus, block *types.Block) error {
  91         for _, tx := range block.Transactions {
  92                 for _, input := range tx.Inputs {
  93                         if input.TypedInput.InputType() != types.VetoInputType {
  94                                 continue
  95                         }
  96
  97                         outputID, err := input.SpentOutputID()
  98                         if err != nil {
  99                                 return err
 100                         }
 101
 102                         result := db.Model(&orm.Utxo{}).Where(&orm.Utxo{OutputID: outputID.String()}).Update("veto_height", block.Height)
 103                         if err := result.Error; err != nil {
 104                                 return err
 105                         } else if result.RowsAffected != 1 {
 106                                 return ErrInconsistentDB
 107                         }
 108                 }
 109
 110                 for i, output := range tx.Outputs {
 111                         voteOutput, ok := output.TypedOutput.(*types.VoteOutput)
 112                         if !ok {
 113                                 continue
 114                         }
 115
 116                         utxo := &orm.Utxo{
 117                                 Xpub:        hex.EncodeToString(voteOutput.Vote),
 118                                 VoteAddress: common.GetAddressFromControlProgram(voteOutput.ControlProgram),
 119                                 VoteHeight:  block.Height,
 120                                 VoteNum:     voteOutput.Amount,
 121                                 OutputID:    tx.OutputID(i).String(),
 122                         }
 123
 124                         if err := db.Save(utxo).Error; err != nil {
 125                                 return err
 126                         }
 127                 }
 128         }
 129
 130         return c.updateChainStatus(db, chainStatus, block)
 131 }
 132
 133 func (c *ChainKeeper) DetachBlock(db *gorm.DB, chainStatus *orm.ChainStatus, block *types.Block) error {
 134         if err := db.Where(&orm.Utxo{VoteHeight: block.Height}).Delete(&orm.Utxo{}).Error; err != nil {
 135                 return err
 136         }
 137
 138         if err := db.Where(&orm.Utxo{VetoHeight: block.Height}).Update("veto_height", 0).Error; err != nil {
 139                 return err
 140         }
 141
 142         return c.updateChainStatus(db, chainStatus, block)
 143 }
 144
 145 func (c *ChainKeeper) initBlockState() error {
 146         block, err := c.node.GetBlockByHeight(0)
 147         if err != nil {
 148                 return errors.Wrap(err, "fail on get genenis block")
 149         }
 150
 151         blockHash := block.Hash()
 152         chainStatus := &orm.ChainStatus{
 153                 BlockHeight: block.Height,
 154                 BlockHash:   blockHash.String(),
 155         }
 156         return c.db.Save(chainStatus).Error
 157 }
 158
 159 func (c *ChainKeeper) updateChainStatus(db *gorm.DB, chainStatus *orm.ChainStatus, block *types.Block) error {
 160         blockHash := block.Hash()
 161         result := db.Model(&orm.ChainStatus{}).Where(chainStatus).Updates(&orm.ChainStatus{
 162                 BlockHeight: block.Height,
 163                 BlockHash:   blockHash.String(),
 164         })
 165         if err := result.Error; err != nil {
 166                 return err
 167         } else if result.RowsAffected != 1 {
 168                 return ErrInconsistentDB
 169         }
 170         return nil
 171 }